Fixed bug in tentacles.
This commit is contained in:
1 parent
4be35fcf9c
commit
2fa9bc8724
22 files changed
+766
-628
No files matched your search
@@ -1,5 +1,9 @@
|
||||
# Repository Guidelines
|
||||
|
||||
## CRITICAL: Init-System Independence
|
||||
|
||||
Cthulhu must never depend on a specific Linux init system or service manager. This applies to the reader, Tentacles, process supervision, installation, packaging, and required tests. Do not require systemd, `systemd-run`, `systemctl`, or a systemd user manager. Use init-independent Linux interfaces and detect the capabilities actually available. Cthulhu and its Tentacles must work on non-systemd distributions, including Artix.
|
||||
|
||||
## Project Structure & Module Organization
|
||||
|
||||
Cthulhu is a Go Linux screen reader. `cmd/` contains executables; `internal/` owns reader subsystems. Embedded audio lives in `assets/sounds/`, Tentacle libraries in `sdk/`, native harnesses in `tests/`, and parity records in `testdata/`. Go tests sit beside implementation files. Arch packaging lives in `distro-packages/Arch-Linux/cthulhu-git/`.
|
||||
|
||||
+1
-1
@@ -55,7 +55,7 @@ Before extending application orchestration, an event loop, native callback, or s
|
||||
|
||||
Preserve event order, GLib/cgo ownership, and native lifetimes. Balance key grabs and releases, recover from dead AT-SPI objects, and clean up on shutdown. Keep core input and focus handling bounded and responsive.
|
||||
|
||||
Tentacles already use an external process protocol: preserve existing host safeguards and author APIs. Keep runnable implementations in external repositories and leave a new installation's catalog empty until repositories are added. Tentacles must not block input, focus, or speech. Treat cgroup-v2 containment as resource control, not a same-user security sandbox. Preserve HTTPS artifact and SHA-256 verification before atomic package installation.
|
||||
Tentacles already use an external process protocol: preserve existing host safeguards and author APIs. Keep runnable implementations in external repositories and leave a new installation's catalog empty until repositories are added. Tentacles must not block input, focus, or speech. Use init-independent process supervision for Tentacles by default, with optional explicitly configured cgroup-v2 resource limits. Do not require a particular init system, a service manager, or administrator setup for normal Tentacle use. Preserve descendant cleanup on disable and reader exit. Treat cgroup-v2 containment as resource control, not a same-user security sandbox. Preserve HTTPS artifact and SHA-256 verification before atomic package installation.
|
||||
|
||||
## Behavioral References and User-Facing Messages
|
||||
|
||||
|
||||
@@ -104,3 +104,24 @@ arrows, selection and Backspace still report editing feedback. Re-enable it
|
||||
and check capital/lowercase voices and password exclusion. Automated tests
|
||||
cover queued speech and Braille; audible speech and a physical Braille display
|
||||
still require live verification.
|
||||
|
||||
## Tentacle process supervision
|
||||
|
||||
Installed Tentacles run as the current user and require no administrator setup or
|
||||
particular init system. Each session has a private Linux subreaper helper. The
|
||||
helper watches a reader-owned lifetime pipe and terminates and reaps its child
|
||||
and detached descendants when the session ends, including abrupt reader exit.
|
||||
Installation and enabling report success only after registration, keyboard
|
||||
shortcut setup and activation succeed. A failed start leaves the installed
|
||||
package disabled, with an error in Preferences; enabling it again retries.
|
||||
|
||||
Only install Tentacles from publishers you trust. They inherit your account's
|
||||
file and network access. Protocol sizes, request deadlines and host admission
|
||||
limits always apply. CPU, memory and task limits require an explicitly configured,
|
||||
writable cgroup v2 parent with the `cpu`, `memory` and `pids` controllers enabled.
|
||||
Set `CTHULHU_TENTACLE_CGROUP_PARENT` or pass
|
||||
`--tentacle-cgroup-parent /absolute/delegated/parent` to opt in. Provisioning that
|
||||
parent is an administrator or session-launcher responsibility, independent of
|
||||
any init system. An invalid explicit parent fails startup rather than silently
|
||||
ignoring the requested limits. Without that configuration, supervision works
|
||||
without kernel resource quotas; it does not provide a security sandbox.
|
||||
+4
-6
@@ -57,9 +57,9 @@ func run(ctx context.Context, arguments []string, stdout, stderr io.Writer) (res
|
||||
options := flag.NewFlagSet("cthulhu", flag.ContinueOnError)
|
||||
options.SetOutput(stderr)
|
||||
tentacleFile := options.String("tentacles", "", "explicit local tentacle launch list (experimental JSON)")
|
||||
tentacleParent := options.String("tentacle-cgroup-parent", "auto", "delegated cgroup v2 parent for local tentacles, or auto")
|
||||
tentacleParent := options.String("tentacle-cgroup-parent", "auto", "optional delegated cgroup v2 parent; auto uses CTHULHU_TENTACLE_CGROUP_PARENT or process supervision")
|
||||
noTentacles := options.Bool("no-tentacles", false, "start without tentacles, ignoring saved tentacles and the local launch list")
|
||||
tentacleGuardian := options.String("tentacle-guardian", "", "run the internal tentacle lifetime helper")
|
||||
tentacleGuardian := options.Bool("tentacle-guardian", false, "run the internal tentacle lifetime helper")
|
||||
showVersion := options.Bool("version", false, "print version and exit")
|
||||
shortVersion := options.Bool("v", false, "print version and exit")
|
||||
replace := options.Bool("replace", false, "stop the running Cthulhu instance before starting")
|
||||
@@ -131,10 +131,8 @@ func run(ctx context.Context, arguments []string, stdout, stderr io.Writer) (res
|
||||
if *preferencesUI {
|
||||
return preferences.Run(ctx, os.Stdin, stdout)
|
||||
}
|
||||
if *tentacleGuardian != "" {
|
||||
stopInput := context.AfterFunc(ctx, func() { os.Stdin.Close() })
|
||||
defer stopInput()
|
||||
return tentacle.RunGuardian(os.Stdin, stdout, *tentacleGuardian)
|
||||
if *tentacleGuardian {
|
||||
return tentacle.RunSupervisor(ctx)
|
||||
}
|
||||
if *ocrConfirm {
|
||||
return ocrconfirm.Run(os.Stdin, stdout)
|
||||
|
||||
@@ -4,7 +4,7 @@
|
||||
pkgbase=cthulhu-git
|
||||
pkgname=cthulhu-git
|
||||
_pkgname=cthulhu
|
||||
pkgver=2026.10.14.r6.gb2f558c
|
||||
pkgver=2026.10.15.r26.g4be35fc
|
||||
pkgrel=1
|
||||
pkgdesc="Desktop-agnostic screen reader with speech, Braille, and Tentacle support"
|
||||
url="https://git.stormux.org/storm/cthulhu"
|
||||
@@ -36,9 +36,8 @@ depends=(
|
||||
libxkbcommon
|
||||
xclip
|
||||
|
||||
# Private D-Bus services and contained Tentacle processes
|
||||
# Private D-Bus services
|
||||
dbus
|
||||
systemd
|
||||
)
|
||||
optdepends=(
|
||||
'git: Discover and update Tentacle repositories'
|
||||
|
||||
@@ -52,13 +52,13 @@ func (worker *preferencesWorker) manage_tentacle(ctx context.Context, request pr
|
||||
response.Message = "Tentacle status."
|
||||
}
|
||||
if request.Tentacle.Action == "retry" {
|
||||
response.Message = "Tentacle retry requested."
|
||||
response.Message = "Tentacle is active."
|
||||
}
|
||||
switch request.Tentacle.Action {
|
||||
case "install":
|
||||
response.Message = "Tentacle installed and saved. Enable it when you are ready."
|
||||
case "install-enable":
|
||||
response.Message = "Tentacle installed and enabled; runtime status is shown separately."
|
||||
response.Message = "Tentacle installed and active."
|
||||
case "repository-add":
|
||||
response.Message = "Repository added. Its tentacles are listed in the table."
|
||||
case "repository-refresh":
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
"compress/gzip"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
@@ -18,6 +19,8 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.stormux.org/storm/cthulhu/internal/config"
|
||||
"git.stormux.org/storm/cthulhu/internal/preferences"
|
||||
"git.stormux.org/storm/cthulhu/internal/tentacle"
|
||||
)
|
||||
|
||||
@@ -76,10 +79,7 @@ func TestRepositoryPackageInstallEnableDisableRemoveAndOfflineRestart(t *testing
|
||||
test_repository_package_install(t, false)
|
||||
}
|
||||
|
||||
func TestNativeRepositoryPackageInstall(t *testing.T) {
|
||||
if os.Getenv("CTHULHU_TENTACLE_MANAGED") != "1" {
|
||||
t.Skip("requires a systemd user manager and cgroup v2")
|
||||
}
|
||||
func TestSupervisedRepositoryPackageInstall(t *testing.T) {
|
||||
test_repository_package_install(t, true)
|
||||
}
|
||||
|
||||
@@ -211,3 +211,70 @@ func TestRepositoryPackageInstallFailureCancellationAndChangedCatalog(t *testing
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestInstallEnableReportsActivationAndDisablesFailedStarts(t *testing.T) {
|
||||
for _, problem := range []string{"none", "manager", "program", "registration", "bindings"} {
|
||||
t.Run(problem, func(t *testing.T) {
|
||||
repository, data := install_fixture(t)
|
||||
set_install_transport(t, func(*http.Request) (*http.Response, error) {
|
||||
return &http.Response{StatusCode: 200, Body: io.NopCloser(bytes.NewReader(data)), Header: make(http.Header)}, nil
|
||||
})
|
||||
registryPath := filepath.Join(t.TempDir(), "tentacles.toml")
|
||||
owner, bindings := managed_tentacle_fixture(t, registryPath, false)
|
||||
if problem == "manager" {
|
||||
owner.newManager = func() (*tentacle.Manager, error) { return nil, errors.New("fixture containment unavailable") }
|
||||
}
|
||||
bindings.reject = problem == "bindings"
|
||||
owner.preparePackage = func(ctx context.Context, root string, repo tentacle.Repository, id string) (*tentacle.PreparedPackage, error) {
|
||||
prepared, err := tentacle.PreparePackage(ctx, root, repo, id)
|
||||
if err == nil {
|
||||
if problem == "program" {
|
||||
if err := os.Remove(prepared.Entry.Program[0]); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if problem == "registration" {
|
||||
prepared.Entry.Program[len(prepared.Entry.Program)-1] = filepath.Join(t.TempDir(), "bad-startup")
|
||||
}
|
||||
}
|
||||
return prepared, err
|
||||
}
|
||||
owner.fetchRepository = func(context.Context, string) (tentacle.Repository, error) { return repository, nil }
|
||||
if err := owner.Start(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := owner.Manage(context.Background(), tentacle.ManagementRequest{Action: "repository-add", URL: repository.URL}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
worker := newPreferencesWorker(owner.logger, config.DefaultSettings(), nil)
|
||||
worker.tentacles = owner
|
||||
response := worker.manage_tentacle(context.Background(), preferences.Request{Tentacle: &tentacle.ManagementRequest{Action: "install-enable", URL: repository.URL, ID: "echo"}})
|
||||
if response.Accepted != (problem == "none") {
|
||||
t.Fatalf("unexpected installation response: %+v", response)
|
||||
}
|
||||
if response.Tentacles == nil || len(response.Tentacles.Entries) != 1 {
|
||||
t.Fatalf("missing installed state: %+v", response)
|
||||
}
|
||||
status := response.Tentacles.Entries[0]
|
||||
if problem == "none" {
|
||||
if status.State != "active" || !status.Entry.Enabled || !strings.Contains(response.Message, "active") {
|
||||
t.Fatalf("success before activation: %+v %s", status, response.Message)
|
||||
}
|
||||
} else {
|
||||
if status.Entry.Enabled || status.Message == "" || strings.Contains(response.Message, "installed and") {
|
||||
t.Fatalf("failed startup reported success: %+v %s", status, response.Message)
|
||||
}
|
||||
if len(bindings.TentacleBindings()) != 0 {
|
||||
t.Fatal("failed startup retained bindings")
|
||||
}
|
||||
if err := owner.Manage(context.Background(), tentacle.ManagementRequest{Action: "enable", ID: status.Entry.ID}); err == nil {
|
||||
t.Fatal("failed enable reported success")
|
||||
}
|
||||
}
|
||||
disk, err := (&tentacle.Registry{Path: registryPath}).Load()
|
||||
if err != nil || len(disk) != 1 || disk[0].Enabled != (problem == "none") {
|
||||
t.Fatalf("incorrect persisted enablement: %+v %v", disk, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -7,7 +7,6 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"reflect"
|
||||
"slices"
|
||||
"strings"
|
||||
@@ -25,6 +24,14 @@ type tentacleManagementCall struct {
|
||||
packageFiles *tentacle.PreparedPackage
|
||||
}
|
||||
|
||||
// A startup request is owned by the lifecycle worker until activation or failure.
|
||||
// Cancellation only wakes that worker; it never mutates registry or session state.
|
||||
type tentacleStartupCall struct {
|
||||
call tentacleManagementCall
|
||||
saveError error
|
||||
stop func() bool
|
||||
}
|
||||
|
||||
func (runtime *tentacleRuntime) Snapshot() tentacle.Snapshot {
|
||||
if runtime == nil {
|
||||
return tentacle.Snapshot{Suppressed: true, Message: "Tentacle management is unavailable for this run."}
|
||||
@@ -201,10 +208,9 @@ func (runtime *tentacleRuntime) Manage(ctx context.Context, request tentacle.Man
|
||||
default:
|
||||
return tentacle.ErrBusy
|
||||
}
|
||||
if call.packageFiles != nil {
|
||||
// Once queued, wait for the lifecycle owner's decision before releasing
|
||||
// files. Returning on caller cancellation here could delete a committed
|
||||
// package. The owner checks cancellation before saving the registry.
|
||||
// Once queued, wait for the lifecycle owner's decision. Startup requests
|
||||
// wake the owner on cancellation so rollback completes before the response;
|
||||
// package files remain owned until the registry decision is known.
|
||||
select {
|
||||
case err := <-call.result:
|
||||
return err
|
||||
@@ -212,15 +218,6 @@ func (runtime *tentacleRuntime) Manage(ctx context.Context, request tentacle.Man
|
||||
return tentacle.ErrUnavailable
|
||||
}
|
||||
}
|
||||
select {
|
||||
case err := <-call.result:
|
||||
return err
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-runtime.done:
|
||||
return tentacle.ErrUnavailable
|
||||
}
|
||||
}
|
||||
|
||||
func change_tentacle_entries(entries []tentacle.Entry, request tentacle.ManagementRequest) ([]tentacle.Entry, error) {
|
||||
next := tentacle.CloneEntries(entries)
|
||||
@@ -270,6 +267,7 @@ func (runtime *tentacleRuntime) run_managed(ctx context.Context) error {
|
||||
pending := make(map[string]bool)
|
||||
sessions := make(map[string]*tentacle.Session)
|
||||
problems := make(map[string]string)
|
||||
startups := make(map[string]tentacleStartupCall)
|
||||
removedPackages := make(map[string]tentacle.Entry)
|
||||
for _, entry := range entries {
|
||||
pending[entry.ID] = entry.Enabled
|
||||
@@ -282,8 +280,6 @@ func (runtime *tentacleRuntime) run_managed(ctx context.Context) error {
|
||||
}
|
||||
}
|
||||
var manager *tentacle.Manager
|
||||
var delegation *tentacle.Delegation
|
||||
var serviceDone <-chan struct{}
|
||||
var current *tentacle.Session
|
||||
var ready <-chan struct{}
|
||||
var active []*tentacle.Session
|
||||
@@ -338,13 +334,10 @@ func (runtime *tentacleRuntime) run_managed(ctx context.Context) error {
|
||||
if runtime.readerContext.Err() != nil {
|
||||
bindingError = nil
|
||||
}
|
||||
var managerError, serviceError error
|
||||
var managerError error
|
||||
if manager != nil {
|
||||
managerError = manager.Shutdown(cleanup)
|
||||
}
|
||||
if delegation != nil {
|
||||
serviceError = delegation.Close(cleanup)
|
||||
}
|
||||
for _, entry := range removedPackages {
|
||||
if managerError == nil {
|
||||
if err := tentacle.RemovePackage(tentacle.PackageRoot(runtime.registry.Path), entry); err != nil {
|
||||
@@ -356,7 +349,7 @@ func (runtime *tentacleRuntime) run_managed(ctx context.Context) error {
|
||||
<-done
|
||||
}
|
||||
runtime.results.Wait()
|
||||
runtime.stopError = errors.Join(bindingError, managerError, serviceError)
|
||||
runtime.stopError = errors.Join(bindingError, managerError)
|
||||
if runtime.stopError != nil && runtime.readerContext.Err() == nil {
|
||||
select {
|
||||
case runtime.failures <- runtime.stopError:
|
||||
@@ -372,34 +365,7 @@ func (runtime *tentacleRuntime) run_managed(ctx context.Context) error {
|
||||
return errors.New("native bindings are unavailable")
|
||||
}
|
||||
var err error
|
||||
if runtime.newManager != nil {
|
||||
manager, err = runtime.newManager()
|
||||
return err
|
||||
}
|
||||
parent := runtime.parent
|
||||
if parent == "" || parent == "auto" {
|
||||
program := runtime.guardianProgram
|
||||
if len(program) == 0 {
|
||||
executable, err := os.Executable()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
program = []string{executable, "--tentacle-guardian"}
|
||||
}
|
||||
if delegation == nil {
|
||||
delegation, err = tentacle.StartDelegation(ctx, program)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
serviceDone = delegation.Done()
|
||||
}
|
||||
parent = delegation.Parent
|
||||
}
|
||||
if delegation != nil {
|
||||
manager, err = tentacle.NewDelegatedManager(delegation)
|
||||
} else {
|
||||
manager, err = tentacle.NewContainedManager(tentacle.DefaultLimits(), nil, parent, tentacle.DefaultResourceLimits())
|
||||
}
|
||||
manager, err = runtime.makeManager()
|
||||
return err
|
||||
}
|
||||
revoke := func(id string) {
|
||||
@@ -408,10 +374,58 @@ func (runtime *tentacleRuntime) run_managed(ctx context.Context) error {
|
||||
runtime.output.RetireRevoked()
|
||||
}
|
||||
}
|
||||
replyStartup := func(id string, err error) {
|
||||
startup, exists := startups[id]
|
||||
if !exists {
|
||||
return
|
||||
}
|
||||
delete(startups, id)
|
||||
startup.stop()
|
||||
if err == nil {
|
||||
err = startup.saveError
|
||||
}
|
||||
setSnapshot()
|
||||
startup.call.result <- err
|
||||
}
|
||||
failStartup := func(id, message string) {
|
||||
problems[id] = message
|
||||
if _, exists := startups[id]; !exists {
|
||||
return
|
||||
}
|
||||
revoke(id)
|
||||
if current != nil && current.TentacleID() == id {
|
||||
current, ready = nil, nil
|
||||
}
|
||||
delete(pending, id)
|
||||
next := tentacle.CloneEntries(entries)
|
||||
for index := range next {
|
||||
if next[index].ID == id {
|
||||
next[index].Enabled = false
|
||||
}
|
||||
}
|
||||
err := runtime.registry.Save(next)
|
||||
if err == nil || errors.Is(err, tentacle.ErrRegistrySync) {
|
||||
entries = next
|
||||
}
|
||||
if err != nil {
|
||||
message += " The disabled state could not be saved durably: " + err.Error()
|
||||
}
|
||||
replyStartup(id, errors.New(message))
|
||||
}
|
||||
defer func() {
|
||||
for id := range startups {
|
||||
failStartup(id, "Tentacle startup was cancelled before activation.")
|
||||
}
|
||||
}()
|
||||
for {
|
||||
if ctx.Err() != nil {
|
||||
return nil
|
||||
}
|
||||
for id, startup := range startups {
|
||||
if startup.call.ctx.Err() != nil {
|
||||
failStartup(id, "Tentacle startup was cancelled before activation.")
|
||||
}
|
||||
}
|
||||
// Reconcile revoked sessions before staging new native grabs.
|
||||
retained := active[:0]
|
||||
for _, session := range active {
|
||||
@@ -474,7 +488,7 @@ func (runtime *tentacleRuntime) run_managed(ctx context.Context) error {
|
||||
session, err = manager.Start(ctx, entry.Spec())
|
||||
}
|
||||
if err != nil {
|
||||
problems[entry.ID] = "Tentacle could not start; check resource containment and its program, then retry."
|
||||
failStartup(entry.ID, "Tentacle could not start; check resource containment and its program, then enable it again.")
|
||||
runtime.logger.Warn("start configured tentacle", "tentacle", entry.ID, "error", err)
|
||||
runtime.report("A tentacle could not start.")
|
||||
notify()
|
||||
@@ -517,10 +531,6 @@ func (runtime *tentacleRuntime) run_managed(ctx context.Context) error {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil
|
||||
case <-serviceDone:
|
||||
// A lost containment owner invalidates every session. Do not relaunch
|
||||
// into a disappearing cgroup or run uncontained.
|
||||
return tentacle.ErrContainment
|
||||
case <-wake:
|
||||
case call := <-runtime.requests:
|
||||
if err := call.ctx.Err(); err != nil {
|
||||
@@ -557,6 +567,10 @@ func (runtime *tentacleRuntime) run_managed(ctx context.Context) error {
|
||||
continue
|
||||
}
|
||||
}
|
||||
if _, starting := startups[call.request.ID]; starting && call.request.Action != "disable" && call.request.Action != "remove" {
|
||||
call.result <- tentacle.ErrBusy
|
||||
continue
|
||||
}
|
||||
next, err := change_tentacle_entries(entries, call.request)
|
||||
if err == nil && call.request.Action != "retry" && reflect.DeepEqual(next, entries) {
|
||||
call.result <- nil
|
||||
@@ -567,6 +581,7 @@ func (runtime *tentacleRuntime) run_managed(ctx context.Context) error {
|
||||
}
|
||||
if err == nil || errors.Is(err, tentacle.ErrRegistrySync) {
|
||||
id := call.request.ID
|
||||
replyStartup(id, errors.New("Tentacle startup was cancelled by another operation."))
|
||||
if call.packageFiles != nil {
|
||||
call.packageFiles.Retain()
|
||||
}
|
||||
@@ -578,6 +593,9 @@ func (runtime *tentacleRuntime) run_managed(ctx context.Context) error {
|
||||
}
|
||||
}
|
||||
revoke(id)
|
||||
if current != nil && current.TentacleID() == id {
|
||||
current, ready = nil, nil
|
||||
}
|
||||
entries = next
|
||||
delete(problems, id)
|
||||
delete(pending, id)
|
||||
@@ -590,12 +608,21 @@ func (runtime *tentacleRuntime) run_managed(ctx context.Context) error {
|
||||
stillActive := slices.DeleteFunc(slices.Clone(active), func(session *tentacle.Session) bool { state, _ := session.Status(); return state != tentacle.Active })
|
||||
runtime.publishCommands(stillActive)
|
||||
setSnapshot()
|
||||
if pending[id] {
|
||||
startups[id] = tentacleStartupCall{call: call, saveError: err, stop: context.AfterFunc(call.ctx, notify)}
|
||||
continue
|
||||
}
|
||||
}
|
||||
call.result <- err
|
||||
case <-ready:
|
||||
session := current
|
||||
current, ready = nil, nil
|
||||
if state, _ := session.Status(); state != tentacle.Ready {
|
||||
failStartup(session.TentacleID(), "Tentacle failed during startup; check its program and resource limits, then enable it again.")
|
||||
continue
|
||||
}
|
||||
if startup, exists := startups[session.TentacleID()]; exists && startup.call.ctx.Err() != nil {
|
||||
failStartup(session.TentacleID(), "Tentacle startup was cancelled before activation.")
|
||||
continue
|
||||
}
|
||||
candidate := append(slices.Clone(active), session)
|
||||
@@ -603,6 +630,7 @@ func (runtime *tentacleRuntime) run_managed(ctx context.Context) error {
|
||||
session.Disable()
|
||||
runtime.logger.Warn("configured tentacle bindings rejected", "tentacle", session.TentacleID(), "error", err)
|
||||
runtime.report("A tentacle could not acquire its keyboard shortcuts.")
|
||||
failStartup(session.TentacleID(), "Tentacle could not acquire its keyboard shortcuts; enable it again after resolving the conflict.")
|
||||
continue
|
||||
}
|
||||
if err := session.Activate(); err != nil {
|
||||
@@ -614,10 +642,12 @@ func (runtime *tentacleRuntime) run_managed(ctx context.Context) error {
|
||||
}
|
||||
return err
|
||||
}
|
||||
failStartup(session.TentacleID(), "Tentacle stopped before activation; enable it again to retry.")
|
||||
continue
|
||||
}
|
||||
active = candidate
|
||||
runtime.publishCommands(active)
|
||||
replyStartup(session.TentacleID(), nil)
|
||||
runtime.logger.Info("configured tentacle active", "tentacle", session.TentacleID(), "session", session.Identity())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -21,6 +21,13 @@ func managed_tentacle_fixture(t *testing.T, path string, contained bool) (*tenta
|
||||
runtime := newTentacleRuntime(slog.New(slog.DiscardHandler), bindings, "", "auto")
|
||||
runtime.registry = &tentacle.Registry{Path: path}
|
||||
runtime.requests = make(chan tentacleManagementCall, 16)
|
||||
if contained {
|
||||
executable, err := os.Executable()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
runtime.guardianProgram = []string{executable, "-test.run=^TestManagedTentacleGuardianPeer$", "--", "managed-guardian"}
|
||||
}
|
||||
if !contained {
|
||||
runtime.newManager = func() (*tentacle.Manager, error) { return tentacle.NewManager(tentacle.DefaultLimits(), nil) }
|
||||
}
|
||||
@@ -219,19 +226,16 @@ func TestPreferencesTentacleOperationsAreGlobalAndSeparate(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestManagedTentacleGuardianPeer(t *testing.T) {
|
||||
if len(os.Args) < 3 || os.Args[len(os.Args)-2] != "managed-guardian" {
|
||||
if len(os.Args) < 2 || os.Args[len(os.Args)-1] != "managed-guardian" {
|
||||
return
|
||||
}
|
||||
if err := tentacle.RunGuardian(os.Stdin, os.Stdout, os.Args[len(os.Args)-1]); err != nil {
|
||||
if err := tentacle.RunSupervisor(context.Background()); err != nil {
|
||||
os.Exit(1)
|
||||
}
|
||||
os.Exit(0)
|
||||
}
|
||||
|
||||
func TestNativeManagedTentacle(t *testing.T) {
|
||||
if os.Getenv("CTHULHU_TENTACLE_MANAGED") != "1" {
|
||||
t.Skip("requires a systemd user manager and cgroup v2")
|
||||
}
|
||||
func TestManagedSupervisedTentacle(t *testing.T) {
|
||||
runtime, bindings := managed_tentacle_fixture(t, filepath.Join(t.TempDir(), "tentacles.toml"), true)
|
||||
readerGroup, err := os.ReadFile("/proc/self/cgroup")
|
||||
if err != nil {
|
||||
@@ -258,13 +262,17 @@ func TestNativeManagedTentacle(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var aggregateGroup string
|
||||
if os.Getenv("CTHULHU_TENTACLE_CGROUP_PARENT") != "" {
|
||||
childGroup := filepath.Join("/sys/fs/cgroup", strings.TrimSpace(strings.TrimPrefix(string(membership), "0::")))
|
||||
aggregateGroup = filepath.Dir(childGroup)
|
||||
for name, want := range map[string]string{"cpu.max": "25000 100000", "memory.max": "268435456", "pids.max": "64"} {
|
||||
value, err := os.ReadFile(filepath.Join(childGroup, name))
|
||||
if err != nil || strings.TrimSpace(string(value)) != want {
|
||||
t.Fatalf("%s: %q %v", name, value, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
afterGroup, err := os.ReadFile("/proc/self/cgroup")
|
||||
if err != nil || string(readerGroup) != string(afterGroup) {
|
||||
t.Fatal("management moved the reader")
|
||||
@@ -278,6 +286,54 @@ func TestNativeManagedTentacle(t *testing.T) {
|
||||
if err := runtime.Stop(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
serviceGroup := filepath.Dir(filepath.Dir(childGroup))
|
||||
waitTentacle(t, func() bool { _, err := os.Stat(serviceGroup); return os.IsNotExist(err) })
|
||||
if aggregateGroup != "" {
|
||||
waitTentacle(t, func() bool { _, err := os.Stat(aggregateGroup); return os.IsNotExist(err) })
|
||||
}
|
||||
}
|
||||
|
||||
func TestManagedTentacleCancelledStartupAndConcurrentDisable(t *testing.T) {
|
||||
for _, operation := range []string{"cancel", "disable", "remove"} {
|
||||
t.Run(operation, func(t *testing.T) {
|
||||
registryPath := filepath.Join(t.TempDir(), "tentacles.toml")
|
||||
owner, bindings := managed_tentacle_fixture(t, registryPath, false)
|
||||
if err := owner.Start(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
entry := probe_entry(t, "probe", "stall")
|
||||
manage_tentacle(t, owner, "add", entry.ID, &entry)
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
result := make(chan error, 1)
|
||||
go func() { result <- owner.Manage(ctx, tentacle.ManagementRequest{Action: "enable", ID: entry.ID}) }()
|
||||
wait_managed_state(t, owner, entry.ID, "starting")
|
||||
if operation == "cancel" {
|
||||
cancel()
|
||||
} else {
|
||||
manage_tentacle(t, owner, operation, entry.ID, nil)
|
||||
}
|
||||
select {
|
||||
case err := <-result:
|
||||
if err == nil {
|
||||
t.Fatal("cancelled startup reported success")
|
||||
}
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("startup cancellation did not complete")
|
||||
}
|
||||
snapshot := owner.Snapshot()
|
||||
if operation == "remove" {
|
||||
if len(snapshot.Entries) != 0 {
|
||||
t.Fatal(snapshot)
|
||||
}
|
||||
} else if len(snapshot.Entries) != 1 || snapshot.Entries[0].Entry.Enabled {
|
||||
t.Fatal(snapshot)
|
||||
}
|
||||
if len(bindings.TentacleBindings()) != 0 {
|
||||
t.Fatal("cancelled startup published bindings")
|
||||
}
|
||||
disk, err := (&tentacle.Registry{Path: registryPath}).Load()
|
||||
if err != nil || len(disk) > 0 && disk[0].Enabled {
|
||||
t.Fatalf("cancelled startup remained enabled: %+v %v", disk, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
+18
-12
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"os"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
@@ -130,18 +131,7 @@ func (runtime *tentacleRuntime) run(ctx context.Context) error {
|
||||
if err != nil || len(specs) == 0 {
|
||||
return err
|
||||
}
|
||||
var manager *tentacle.Manager
|
||||
if runtime.newManager != nil {
|
||||
manager, err = runtime.newManager()
|
||||
} else {
|
||||
parent := runtime.parent
|
||||
if parent == "auto" {
|
||||
parent, err = tentacle.DelegatedParent()
|
||||
}
|
||||
if err == nil {
|
||||
manager, err = tentacle.NewContainedManager(tentacle.DefaultLimits(), nil, parent, tentacle.DefaultResourceLimits())
|
||||
}
|
||||
}
|
||||
manager, err := runtime.makeManager()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -422,3 +412,19 @@ func (runtime *tentacleRuntime) report(message string) {
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
// makeManager runs on the lifecycle worker, never on input or GTK callbacks.
|
||||
func (runtime *tentacleRuntime) makeManager() (*tentacle.Manager, error) {
|
||||
if runtime.newManager != nil {
|
||||
return runtime.newManager()
|
||||
}
|
||||
program := runtime.guardianProgram
|
||||
if len(program) == 0 {
|
||||
executable, err := os.Executable()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
program = []string{executable, "--tentacle-guardian"}
|
||||
}
|
||||
return tentacle.NewSupervisedManager(program, runtime.parent)
|
||||
}
|
||||
@@ -26,6 +26,9 @@ func TestTentacleReaderPeer(t *testing.T) {
|
||||
}
|
||||
marker := os.Args[len(os.Args)-1]
|
||||
presentation := strings.HasPrefix(filepath.Base(marker), "present")
|
||||
if filepath.Base(marker) == "bad-startup" {
|
||||
os.Exit(2)
|
||||
}
|
||||
if filepath.Base(marker) == "stall" {
|
||||
time.Sleep(time.Minute)
|
||||
os.Exit(0)
|
||||
@@ -110,6 +113,9 @@ func readerTentacleFixture(t *testing.T, mode string, contained bool) (*tentacle
|
||||
}
|
||||
bindings := &tentacleBindingRecorder{reject: mode == "reject"}
|
||||
runtime := newTentacleRuntime(slog.New(slog.DiscardHandler), bindings, path, "auto")
|
||||
if contained {
|
||||
runtime.guardianProgram = []string{executable, "-test.run=^TestManagedTentacleGuardianPeer$", "--", "managed-guardian"}
|
||||
}
|
||||
if !contained {
|
||||
runtime.newManager = func() (*tentacle.Manager, error) { return tentacle.NewManager(tentacle.DefaultLimits(), nil) }
|
||||
}
|
||||
|
||||
@@ -1,161 +0,0 @@
|
||||
// Copyright (c) 2026 Storm Dragon
|
||||
// SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
package tentacle
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Delegation owns a transient user service, not the reader process. Its small
|
||||
// guardian exits on pipe EOF, including reader death; systemd then kills the
|
||||
// whole service subtree. No service or example is installed or enabled on disk.
|
||||
type Delegation struct {
|
||||
Parent string
|
||||
input io.WriteCloser
|
||||
output io.ReadCloser
|
||||
done chan struct{}
|
||||
waitError error
|
||||
once sync.Once
|
||||
unit string
|
||||
directory string
|
||||
process *os.Process
|
||||
}
|
||||
|
||||
func StartDelegation(ctx context.Context, program []string) (*Delegation, error) {
|
||||
if len(program) == 0 || !filepath.IsAbs(program[0]) {
|
||||
return nil, ErrContainment
|
||||
}
|
||||
var identity [16]byte
|
||||
if _, err := rand.Read(identity[:]); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
unit := fmt.Sprintf("cthulhu-tentacles-%x.service", identity)
|
||||
directory, err := os.MkdirTemp("", "cthulhu-tentacles-")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
keepDirectory := false
|
||||
defer func() {
|
||||
if !keepDirectory {
|
||||
os.Remove(filepath.Join(directory, "launch"))
|
||||
os.Remove(directory)
|
||||
}
|
||||
}()
|
||||
args := []string{"--user", "--quiet", "--wait", "--pipe", "--collect", "--unit=" + unit,
|
||||
"--property=Delegate=cpu memory pids", "--property=DelegateSubgroup=reader",
|
||||
"--property=KillMode=control-group", "--property=TimeoutStopSec=3s"}
|
||||
for _, name := range []string{"DISPLAY", "XAUTHORITY", "DBUS_SESSION_BUS_ADDRESS", "AT_SPI_BUS_ADDRESS",
|
||||
"XDG_RUNTIME_DIR", "XDG_SESSION_TYPE", "XDG_SESSION_DESKTOP", "XDG_DATA_HOME", "XDG_CONFIG_HOME", "XDG_CACHE_HOME",
|
||||
"LANG", "LANGUAGE", "LC_ALL", "SPEECHD_ADDRESS", "SPEECHD_SOCKET"} {
|
||||
if value, ok := os.LookupEnv(name); ok {
|
||||
args = append(args, "--setenv="+name+"="+value)
|
||||
}
|
||||
}
|
||||
args = append(append(args, "--"), program...)
|
||||
args = append(args, filepath.Join(directory, "launch"))
|
||||
command := exec.Command("systemd-run", args...)
|
||||
// No inherited terminal input or unbounded diagnostics. The handshake below
|
||||
// is the only stdout message; failure becomes a bounded reader status.
|
||||
input, err := command.StdinPipe()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
output, err := command.StdoutPipe()
|
||||
if err != nil {
|
||||
input.Close()
|
||||
return nil, err
|
||||
}
|
||||
if err := command.Start(); err != nil {
|
||||
input.Close()
|
||||
output.Close()
|
||||
return nil, err
|
||||
}
|
||||
delegation := &Delegation{input: input, output: output, done: make(chan struct{}), unit: unit, directory: directory, process: command.Process}
|
||||
go func() { delegation.waitError = command.Wait(); close(delegation.done) }()
|
||||
readDone := make(chan error, 1)
|
||||
go func() {
|
||||
scanner := bufio.NewScanner(output)
|
||||
scanner.Buffer(make([]byte, 4096), 4096)
|
||||
if !scanner.Scan() {
|
||||
readDone <- ErrContainment
|
||||
return
|
||||
}
|
||||
parent := scanner.Text()
|
||||
if !strings.HasPrefix(parent, "/sys/fs/cgroup/") || filepath.Clean(parent) != parent || filepath.Base(parent) != unit {
|
||||
readDone <- ErrContainment
|
||||
return
|
||||
}
|
||||
delegation.Parent = parent
|
||||
readDone <- nil
|
||||
}()
|
||||
startup, cancel := context.WithTimeout(ctx, 5*time.Second)
|
||||
defer cancel()
|
||||
select {
|
||||
case err = <-readDone:
|
||||
case <-startup.Done():
|
||||
err = startup.Err()
|
||||
output.Close()
|
||||
<-readDone
|
||||
}
|
||||
if err != nil {
|
||||
cleanup, cancel := context.WithTimeout(context.Background(), 4*time.Second)
|
||||
defer cancel()
|
||||
delegation.Close(cleanup)
|
||||
return nil, fmt.Errorf("%w: could not start delegated user service", ErrContainment)
|
||||
}
|
||||
keepDirectory = true
|
||||
return delegation, nil
|
||||
}
|
||||
|
||||
func (delegation *Delegation) Done() <-chan struct{} { return delegation.done }
|
||||
|
||||
func (delegation *Delegation) Close(ctx context.Context) error {
|
||||
delegation.once.Do(func() { delegation.input.Close() })
|
||||
select {
|
||||
case <-delegation.done:
|
||||
delegation.output.Close()
|
||||
os.Remove(filepath.Join(delegation.directory, "launch"))
|
||||
os.Remove(delegation.directory)
|
||||
return delegation.waitError
|
||||
case <-ctx.Done():
|
||||
// Stop only the unique service this owner created. Never stop another reader.
|
||||
stopCtx, cancel := context.WithTimeout(context.Background(), time.Second)
|
||||
defer cancel()
|
||||
stopError := exec.CommandContext(stopCtx, "systemctl", "--user", "--no-block", "stop", delegation.unit).Run()
|
||||
delegation.output.Close()
|
||||
_ = delegation.process.Kill()
|
||||
select {
|
||||
case <-delegation.done:
|
||||
case <-time.After(time.Second):
|
||||
}
|
||||
return errors.Join(ctx.Err(), stopError)
|
||||
}
|
||||
}
|
||||
|
||||
// RunGuardian is a private helper. stdin is a reader-owned lifetime pipe, never
|
||||
// a command protocol. The guardian has no speech, keyboard, or settings access.
|
||||
func RunGuardian(input io.Reader, output io.Writer, socketPath string) error {
|
||||
parent, err := DelegatedParent()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = serve_launches(input, output, socketPath, parent)
|
||||
// The reader may have died; the guardian also removes its empty private
|
||||
// socket directory after the listener unlinks the socket.
|
||||
if filepath.Base(socketPath) == "launch" && strings.HasPrefix(filepath.Base(filepath.Dir(socketPath)), "cthulhu-tentacles-") {
|
||||
os.Remove(filepath.Dir(socketPath))
|
||||
}
|
||||
return err
|
||||
}
|
||||
@@ -1,99 +0,0 @@
|
||||
package tentacle
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestDelegationHelper(t *testing.T) {
|
||||
if len(os.Args) < 3 {
|
||||
return
|
||||
}
|
||||
mode, path := os.Args[len(os.Args)-2], os.Args[len(os.Args)-1]
|
||||
if mode == "delegation-guardian" {
|
||||
if err := RunGuardian(os.Stdin, os.Stdout, path); err != nil {
|
||||
os.Exit(1)
|
||||
}
|
||||
os.Exit(0)
|
||||
}
|
||||
if mode != "delegation-owner" {
|
||||
return
|
||||
}
|
||||
executable, _ := os.Executable()
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
delegation, err := StartDelegation(ctx, []string{executable, "-test.run=^TestDelegationHelper$", "--", "delegation-guardian"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
manager, err := NewDelegatedManager(delegation)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
session, err := manager.Start(ctx, child_spec(t, "spawn", "spawn"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
select {
|
||||
case <-session.Ready():
|
||||
case <-ctx.Done():
|
||||
t.Fatal(ctx.Err())
|
||||
}
|
||||
if err := session.Activate(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
result, err := session.Invoke(ctx, "echo", nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
value := <-result
|
||||
if value.Err != nil {
|
||||
t.Fatal(value.Err)
|
||||
}
|
||||
var pid int
|
||||
if err := json.Unmarshal(value.Value, &pid); err != nil || pid <= 0 {
|
||||
t.Fatalf("descendant: %s %v", value.Value, err)
|
||||
}
|
||||
if err := os.WriteFile(path, []byte(delegation.Parent), 0600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// Simulate abrupt reader death: no Stop, Shutdown, Close, or deferred cleanup.
|
||||
os.Exit(0)
|
||||
}
|
||||
|
||||
func TestNativeDelegationReaderDeath(t *testing.T) {
|
||||
if os.Getenv("CTHULHU_TENTACLE_MANAGED") != "1" {
|
||||
t.Skip("requires a systemd user manager and cgroup v2")
|
||||
}
|
||||
executable, _ := os.Executable()
|
||||
marker := filepath.Join(t.TempDir(), "group")
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
|
||||
defer cancel()
|
||||
command := exec.CommandContext(ctx, executable, "-test.run=^TestDelegationHelper$", "--", "delegation-owner", marker)
|
||||
if output, err := command.CombinedOutput(); err != nil {
|
||||
t.Fatalf("owner: %v %s", err, output)
|
||||
}
|
||||
parent, err := os.ReadFile(marker)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
timer := time.NewTimer(6 * time.Second)
|
||||
defer timer.Stop()
|
||||
ticker := time.NewTicker(10 * time.Millisecond)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
if _, err := os.Stat(string(parent)); os.IsNotExist(err) {
|
||||
return
|
||||
}
|
||||
select {
|
||||
case <-ticker.C:
|
||||
case <-timer.C:
|
||||
t.Fatal("reader death left its service/descendants alive")
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,203 +0,0 @@
|
||||
// Copyright (c) 2026 Storm Dragon
|
||||
// SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
package tentacle
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"syscall"
|
||||
"time"
|
||||
)
|
||||
|
||||
type launchReply struct {
|
||||
PID int
|
||||
Errno uint32
|
||||
}
|
||||
|
||||
// launch passes exactly four descriptors: stdin, stdout, stderr and the already
|
||||
// limited cgroup. The service performs clone3 inside its delegated tree. No
|
||||
// shell, environment update, or settings operation is exposed by this protocol.
|
||||
func (delegation *Delegation) launch(ctx context.Context, spec Spec, files ...*os.File) (int, <-chan struct{}, func(), error) {
|
||||
connection, err := net.DialUnix("unixpacket", nil, &net.UnixAddr{Name: filepath.Join(delegation.directory, "launch"), Net: "unixpacket"})
|
||||
if err != nil {
|
||||
return 0, nil, nil, err
|
||||
}
|
||||
keep := false
|
||||
defer func() {
|
||||
if !keep {
|
||||
connection.Close()
|
||||
}
|
||||
}()
|
||||
connection.SetDeadline(time.Now().Add(3 * time.Second))
|
||||
stopCancel := context.AfterFunc(ctx, func() { connection.Close() })
|
||||
defer func() {
|
||||
if !keep {
|
||||
stopCancel()
|
||||
}
|
||||
}()
|
||||
data, err := json.Marshal(spec)
|
||||
if err != nil || len(data) > MaxFrameBytes || len(files) != 4 {
|
||||
return 0, nil, nil, ErrLimit
|
||||
}
|
||||
descriptors := make([]int, len(files))
|
||||
for index, file := range files {
|
||||
descriptors[index] = int(file.Fd())
|
||||
}
|
||||
if _, _, err := connection.WriteMsgUnix(data, syscall.UnixRights(descriptors...), nil); err != nil {
|
||||
return 0, nil, nil, err
|
||||
}
|
||||
buffer := make([]byte, 256)
|
||||
count, err := connection.Read(buffer)
|
||||
if err != nil {
|
||||
return 0, nil, nil, err
|
||||
}
|
||||
var reply launchReply
|
||||
if err := json.Unmarshal(buffer[:count], &reply); err != nil {
|
||||
return 0, nil, nil, ErrProtocol
|
||||
}
|
||||
if reply.Errno != 0 {
|
||||
return 0, nil, nil, syscall.Errno(reply.Errno)
|
||||
}
|
||||
if reply.PID <= 0 {
|
||||
return 0, nil, nil, ErrProtocol
|
||||
}
|
||||
connection.SetDeadline(time.Time{})
|
||||
done := make(chan struct{})
|
||||
go func() { defer stopCancel(); var end [1]byte; connection.Read(end[:]); connection.Close(); close(done) }()
|
||||
keep = true
|
||||
return reply.PID, done, func() { connection.Close() }, nil
|
||||
}
|
||||
|
||||
func serve_launches(input io.Reader, output io.Writer, socketPath, parent string) error {
|
||||
listener, err := net.ListenUnix("unixpacket", &net.UnixAddr{Name: socketPath, Net: "unixpacket"})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer listener.Close()
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
var workers sync.WaitGroup
|
||||
capacity := make(chan struct{}, DefaultLimits().Processes)
|
||||
accepted := make(chan struct{})
|
||||
go func() {
|
||||
defer close(accepted)
|
||||
for {
|
||||
connection, err := listener.AcceptUnix()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
select {
|
||||
case capacity <- struct{}{}:
|
||||
workers.Add(1)
|
||||
go func() { defer workers.Done(); defer func() { <-capacity }(); serve_launch(ctx, connection, parent) }()
|
||||
default:
|
||||
connection.Close()
|
||||
}
|
||||
}
|
||||
}()
|
||||
defer func() { cancel(); listener.Close(); <-accepted; workers.Wait() }()
|
||||
if _, err := fmt.Fprintln(output, parent); err != nil {
|
||||
return err
|
||||
}
|
||||
_, err = io.Copy(io.Discard, input)
|
||||
return err
|
||||
}
|
||||
|
||||
func serve_launch(ctx context.Context, connection *net.UnixConn, parent string) {
|
||||
defer connection.Close()
|
||||
stop := context.AfterFunc(ctx, func() { connection.Close() })
|
||||
defer stop()
|
||||
connection.SetDeadline(time.Now().Add(3 * time.Second))
|
||||
data, ancillary := make([]byte, MaxFrameBytes), make([]byte, syscall.CmsgSpace(4*4))
|
||||
count, rightsCount, flags, _, err := connection.ReadMsgUnix(data, ancillary)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
// Always close received descriptors, even when framing or validation fails.
|
||||
var files []*os.File
|
||||
defer func() {
|
||||
for _, file := range files {
|
||||
file.Close()
|
||||
}
|
||||
}()
|
||||
messages, err := syscall.ParseSocketControlMessage(ancillary[:rightsCount])
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
for _, message := range messages {
|
||||
descriptors, err := syscall.ParseUnixRights(&message)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
for _, descriptor := range descriptors {
|
||||
syscall.CloseOnExec(descriptor)
|
||||
files = append(files, os.NewFile(uintptr(descriptor), "tentacle-launch"))
|
||||
}
|
||||
}
|
||||
if flags&(syscall.MSG_TRUNC|syscall.MSG_CTRUNC) != 0 || len(files) != 4 {
|
||||
return
|
||||
}
|
||||
var spec Spec
|
||||
if decode_strict(data[:count], &spec) != nil || validate_spec(spec) != nil {
|
||||
return
|
||||
}
|
||||
var stat syscall.Statfs_t
|
||||
if syscall.Fstatfs(int(files[3].Fd()), &stat) != nil || stat.Type != 0x63677270 {
|
||||
return
|
||||
}
|
||||
groupPath, err := os.Readlink(fmt.Sprintf("/proc/self/fd/%d", files[3].Fd()))
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
relative, err := filepath.Rel(parent, groupPath)
|
||||
parts := strings.Split(relative, string(filepath.Separator))
|
||||
if err != nil || len(parts) != 2 || !strings.HasPrefix(parts[0], "cthulhu-tentacles-") {
|
||||
return
|
||||
}
|
||||
for _, identity := range []string{strings.TrimPrefix(parts[0], "cthulhu-tentacles-"), parts[1]} {
|
||||
if decoded, err := hex.DecodeString(identity); err != nil || len(decoded) != 16 {
|
||||
return
|
||||
}
|
||||
}
|
||||
command := exec.Command(spec.Program[0], spec.Program[1:]...)
|
||||
command.Dir = spec.Directory
|
||||
command.Stdin, command.Stdout, command.Stderr = files[0], files[1], files[2]
|
||||
command.SysProcAttr = &syscall.SysProcAttr{Setpgid: true, UseCgroupFD: true, CgroupFD: int(files[3].Fd())}
|
||||
if err := command.Start(); err != nil {
|
||||
var errno syscall.Errno
|
||||
if !errors.As(err, &errno) {
|
||||
errno = syscall.EIO
|
||||
}
|
||||
reply, _ := json.Marshal(launchReply{Errno: uint32(errno)})
|
||||
connection.Write(reply)
|
||||
return
|
||||
}
|
||||
for _, file := range files {
|
||||
file.Close()
|
||||
}
|
||||
files = nil
|
||||
// A closed launch connection kills the direct child. The reader's cgroup
|
||||
// cleanup or systemd's guardian-exit cleanup covers every descendant.
|
||||
disconnected := make(chan struct{})
|
||||
go func() { defer close(disconnected); var end [1]byte; connection.Read(end[:]); command.Process.Kill() }()
|
||||
reply, _ := json.Marshal(launchReply{PID: command.Process.Pid})
|
||||
if _, err := connection.Write(reply); err != nil {
|
||||
command.Process.Kill()
|
||||
}
|
||||
connection.SetDeadline(time.Time{})
|
||||
_ = command.Wait()
|
||||
connection.Write([]byte{0})
|
||||
connection.Close()
|
||||
<-disconnected
|
||||
}
|
||||
@@ -7,7 +7,6 @@ import (
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"syscall"
|
||||
)
|
||||
|
||||
@@ -58,35 +57,12 @@ func LoadLocalSpecs(path string) ([]Spec, error) {
|
||||
return specs, nil
|
||||
}
|
||||
|
||||
// DelegatedParent enables controllers only within the launcher's explicitly
|
||||
// delegated unit. The reader already lives in its own leaf; it is never moved
|
||||
// here. Other launchers can supply an already configured absolute parent.
|
||||
// DelegatedParent reads an explicitly configured optional cgroup parent.
|
||||
// It does not inspect init-system naming conventions or move the reader.
|
||||
func DelegatedParent() (string, error) {
|
||||
data, err := os.ReadFile("/proc/self/cgroup")
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
parent, err := delegated_parent(string(data))
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if err := write_control(parent, "cgroup.subtree_control", "+cpu +memory +pids"); err != nil {
|
||||
return "", err
|
||||
parent := os.Getenv("CTHULHU_TENTACLE_CGROUP_PARENT")
|
||||
if !filepath.IsAbs(parent) || filepath.Clean(parent) != parent {
|
||||
return "", fmt.Errorf("%w: configure CTHULHU_TENTACLE_CGROUP_PARENT with an absolute delegated cgroup v2 directory", ErrContainment)
|
||||
}
|
||||
return parent, nil
|
||||
}
|
||||
|
||||
func delegated_parent(membership string) (string, error) {
|
||||
for _, line := range strings.Split(membership, "\n") {
|
||||
if !strings.HasPrefix(line, "0::/") {
|
||||
continue
|
||||
}
|
||||
path := strings.TrimPrefix(line, "0::")
|
||||
if filepath.Clean(path) != path || filepath.Base(path) != "reader" ||
|
||||
!strings.HasSuffix(filepath.Dir(path), ".service") {
|
||||
break
|
||||
}
|
||||
return filepath.Join("/sys/fs/cgroup", filepath.Dir(path)), nil
|
||||
}
|
||||
return "", fmt.Errorf("%w: reader requires a delegated service with DelegateSubgroup=reader", ErrContainment)
|
||||
}
|
||||
@@ -8,13 +8,15 @@ import (
|
||||
)
|
||||
|
||||
func TestDelegatedReaderParent(t *testing.T) {
|
||||
parent, err := delegated_parent("0::/user.slice/test.service/reader\n")
|
||||
if err != nil || parent != "/sys/fs/cgroup/user.slice/test.service" {
|
||||
t.Fatalf("%s %v", parent, err)
|
||||
for _, value := range []string{"", "relative", "/sys/fs/cgroup/../outside", "/sys/fs/cgroup/extensions"} {
|
||||
t.Setenv("CTHULHU_TENTACLE_CGROUP_PARENT", value)
|
||||
parent, err := DelegatedParent()
|
||||
if value == "/sys/fs/cgroup/extensions" {
|
||||
if err != nil || parent != value {
|
||||
t.Fatalf("%q: %s %v", value, parent, err)
|
||||
}
|
||||
for _, path := range []string{"0::/", "0::/test.service", "0::/session.scope/reader", "0::/test.service/../reader", "1:cpu:/test.service/reader"} {
|
||||
if _, err := delegated_parent(path); err == nil {
|
||||
t.Fatalf("accepted %q", path)
|
||||
} else if err == nil {
|
||||
t.Fatalf("accepted %q", value)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
@@ -56,22 +57,36 @@ type Manager struct {
|
||||
closed bool
|
||||
coreBindings map[input.KeyboardLayout]input.CommandBindings
|
||||
containment *containment
|
||||
delegation *Delegation
|
||||
supervisorProgram []string
|
||||
cleanupError error
|
||||
}
|
||||
|
||||
// NewDelegatedManager keeps exec inside the delegated service. This is required
|
||||
// when the reader's own cgroup is outside that service's delegation boundary.
|
||||
func NewDelegatedManager(delegation *Delegation) (*Manager, error) {
|
||||
manager, err := NewContainedManager(DefaultLimits(), nil, delegation.Parent, DefaultResourceLimits())
|
||||
// NewSupervisedManager launches a private lifetime helper for each Tentacle.
|
||||
// The default needs no administrator setup. An explicit cgroup parent (or the
|
||||
// CTHULHU_TENTACLE_CGROUP_PARENT environment variable in auto mode) opts into
|
||||
// kernel resource limits; a broken explicit configuration never falls back.
|
||||
func NewSupervisedManager(program []string, parent string) (*Manager, error) {
|
||||
if len(program) == 0 || !filepath.IsAbs(program[0]) {
|
||||
return nil, errors.New("tentacle supervisor must be an absolute program")
|
||||
}
|
||||
if parent == "" || parent == "auto" {
|
||||
parent = os.Getenv("CTHULHU_TENTACLE_CGROUP_PARENT")
|
||||
}
|
||||
var manager *Manager
|
||||
var err error
|
||||
if parent == "" {
|
||||
manager, err = NewManager(DefaultLimits(), nil)
|
||||
} else {
|
||||
manager, err = NewContainedManager(DefaultLimits(), nil, parent, DefaultResourceLimits())
|
||||
}
|
||||
if err == nil {
|
||||
manager.delegation = delegation
|
||||
manager.supervisorProgram = append([]string(nil), program...)
|
||||
}
|
||||
return manager, err
|
||||
}
|
||||
|
||||
// NewManager is the uncontained host for explicit trusted protocol fixtures.
|
||||
// Reader integration must use NewContainedManager; there is no implicit fallback.
|
||||
// NewManager is the plain process-group host for trusted protocol fixtures.
|
||||
// Reader integration uses NewSupervisedManager to clean up detached descendants.
|
||||
func NewManager(limits Limits, overrides []input.BindingOverride) (*Manager, error) {
|
||||
if limits.Processes < 1 || limits.Starting < 1 || limits.Starting > limits.Processes ||
|
||||
limits.PendingPerProcess < 1 || limits.PendingTotal < limits.PendingPerProcess ||
|
||||
|
||||
@@ -5,6 +5,7 @@ package tentacle
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
@@ -96,7 +97,25 @@ func (session *Session) run() {
|
||||
return
|
||||
}
|
||||
command := exec.Command(session.spec.Program[0], session.spec.Program[1:]...)
|
||||
supervised := len(session.manager.supervisorProgram) != 0
|
||||
var lifetime, childLifetime, configuration, childConfiguration *os.File
|
||||
if supervised {
|
||||
childLifetime, lifetime, err = newPipe()
|
||||
if err != nil {
|
||||
session.fail(ErrSupervision)
|
||||
return
|
||||
}
|
||||
childConfiguration, configuration, err = newPipe()
|
||||
if err != nil {
|
||||
session.fail(ErrSupervision)
|
||||
return
|
||||
}
|
||||
program := session.manager.supervisorProgram
|
||||
command = exec.Command(program[0], program[1:]...)
|
||||
command.ExtraFiles = []*os.File{childLifetime, childConfiguration}
|
||||
} else {
|
||||
command.Dir = session.spec.Directory
|
||||
}
|
||||
command.Stdin, command.Stdout, command.Stderr = childInput, childOutput, childDiagnostics
|
||||
command.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
|
||||
var group *cgroup
|
||||
@@ -112,23 +131,13 @@ func (session *Session) run() {
|
||||
command.SysProcAttr.UseCgroupFD = true
|
||||
command.SysProcAttr.CgroupFD = int(group.file.Fd())
|
||||
}
|
||||
var pid int
|
||||
var exited <-chan struct{}
|
||||
if session.manager.delegation != nil {
|
||||
var closeLaunch func()
|
||||
pid, exited, closeLaunch, err = session.manager.delegation.launch(session.ctx, session.spec, childInput, childOutput, childDiagnostics, group.file)
|
||||
if err == nil {
|
||||
defer closeLaunch()
|
||||
}
|
||||
} else {
|
||||
var waitError error
|
||||
exited := make(chan struct{})
|
||||
err = command.Start()
|
||||
if err == nil {
|
||||
pid = command.Process.Pid
|
||||
done := make(chan struct{})
|
||||
exited = done
|
||||
go func() { _ = command.Wait(); close(done) }()
|
||||
}
|
||||
go func() { waitError = command.Wait(); close(exited) }()
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
// Do not retain the command line, environment or child-supplied text.
|
||||
if group != nil {
|
||||
@@ -147,25 +156,52 @@ func (session *Session) run() {
|
||||
childOutput.Close()
|
||||
childDiagnostics.Close()
|
||||
var workers sync.WaitGroup
|
||||
if supervised {
|
||||
childLifetime.Close()
|
||||
childConfiguration.Close()
|
||||
workers.Add(1)
|
||||
go func() {
|
||||
defer workers.Done()
|
||||
_ = json.NewEncoder(configuration).Encode(session.spec)
|
||||
configuration.Close()
|
||||
}()
|
||||
}
|
||||
workers.Add(3)
|
||||
go func() { defer workers.Done(); session.write_requests(input) }()
|
||||
go func() { defer workers.Done(); session.read_replies(output) }()
|
||||
go func() { defer workers.Done(); session.drain_diagnostics(diagnostics) }()
|
||||
defer func() {
|
||||
session.cancel()
|
||||
if supervised {
|
||||
lifetime.Close()
|
||||
configuration.Close()
|
||||
}
|
||||
if group != nil {
|
||||
_ = write_control(group.path, "cgroup.kill", "1")
|
||||
if command.Process != nil {
|
||||
_ = command.Process.Kill()
|
||||
}
|
||||
} else {
|
||||
} else if !supervised {
|
||||
// Explicit trusted-fixture mode only; detached descendants escape.
|
||||
_ = syscall.Kill(-pid, syscall.SIGKILL)
|
||||
_ = syscall.Kill(-command.Process.Pid, syscall.SIGKILL)
|
||||
}
|
||||
input.Close()
|
||||
output.Close()
|
||||
diagnostics.Close()
|
||||
if supervised && group == nil {
|
||||
select {
|
||||
case <-exited:
|
||||
if waitError != nil {
|
||||
cleanupError = ErrSupervision
|
||||
}
|
||||
case <-time.After(DefaultResourceLimits().CleanupTimeout + time.Second):
|
||||
cleanupError = ErrSupervision
|
||||
_ = command.Process.Kill()
|
||||
<-exited
|
||||
}
|
||||
} else {
|
||||
<-exited
|
||||
}
|
||||
workers.Wait()
|
||||
}()
|
||||
|
||||
|
||||
@@ -420,6 +420,16 @@ func TestTentacleChild(t *testing.T) {
|
||||
if !found {
|
||||
return
|
||||
}
|
||||
if mode == "descendant-chain" {
|
||||
executable, _ := os.Executable()
|
||||
child := exec.Command(executable, "-test.run=^TestTentacleChild$", "--", "tentacle-child:descendant")
|
||||
child.SysProcAttr = &syscall.SysProcAttr{Setsid: true}
|
||||
if child.Start() != nil {
|
||||
os.Exit(8)
|
||||
}
|
||||
fmt.Fprintln(os.Stdout, child.Process.Pid)
|
||||
os.Exit(0)
|
||||
}
|
||||
if mode == "descendant" {
|
||||
<-time.After(time.Hour)
|
||||
os.Exit(0)
|
||||
@@ -482,6 +492,9 @@ func TestTentacleChild(t *testing.T) {
|
||||
}
|
||||
reply(registration)
|
||||
case "initialized":
|
||||
if mode == "exit-active" {
|
||||
os.Exit(7)
|
||||
}
|
||||
if mode == "blocked-input" {
|
||||
<-time.After(time.Hour)
|
||||
}
|
||||
@@ -491,10 +504,29 @@ func TestTentacleChild(t *testing.T) {
|
||||
os.Exit(7)
|
||||
case "timeout", "cancel", "disable":
|
||||
continue
|
||||
case "spawn":
|
||||
case "spawn", "spawn-detached", "spawn-chain":
|
||||
executable, _ := os.Executable()
|
||||
child := exec.Command(executable, "-test.run=^TestTentacleChild$", "--", "tentacle-child:descendant")
|
||||
child.Stdout, child.Stderr = os.Stdout, os.Stderr
|
||||
if mode == "spawn-detached" {
|
||||
child.SysProcAttr = &syscall.SysProcAttr{Setsid: true}
|
||||
}
|
||||
if mode == "spawn-chain" {
|
||||
child.Args[len(child.Args)-1] = "tentacle-child:descendant-chain"
|
||||
child.Stdout = nil
|
||||
output, err := child.StdoutPipe()
|
||||
if err != nil || child.Start() != nil {
|
||||
os.Exit(8)
|
||||
}
|
||||
var pid int
|
||||
if _, err := fmt.Fscanln(output, &pid); err != nil {
|
||||
os.Exit(8)
|
||||
}
|
||||
_ = child.Wait()
|
||||
output.Close()
|
||||
reply(pid)
|
||||
continue
|
||||
}
|
||||
if child.Start() != nil {
|
||||
os.Exit(8)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,151 @@
|
||||
// Copyright (c) 2026 Storm Dragon
|
||||
// SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
package tentacle
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"runtime"
|
||||
"strconv"
|
||||
"strings"
|
||||
"syscall"
|
||||
"time"
|
||||
)
|
||||
|
||||
var ErrSupervision = errors.New("tentacle process supervision failed")
|
||||
|
||||
// RunSupervisor is a private per-session helper. Descriptors 0..2 carry only
|
||||
// the Tentacle protocol and diagnostics; 3 is a reader-owned lifetime pipe and
|
||||
// 4 is a bounded Spec. The child never inherits either private descriptor.
|
||||
// A separate subreaper keeps ownership of double-forked/setsid descendants
|
||||
// without making the reader a subreaper for unrelated desktop helpers.
|
||||
func RunSupervisor(ctx context.Context) error {
|
||||
// ExtraFiles arrive in blocking mode. Register the private pipes with Go's
|
||||
// poller so Close interrupts their readers on cancellation and child exit.
|
||||
for _, descriptor := range []int{3, 4} {
|
||||
syscall.CloseOnExec(descriptor)
|
||||
if err := syscall.SetNonblock(descriptor, true); err != nil {
|
||||
return fmt.Errorf("%w: private pipe: %w", ErrSupervision, err)
|
||||
}
|
||||
}
|
||||
lifetime := os.NewFile(3, "tentacle-lifetime")
|
||||
configuration := os.NewFile(4, "tentacle-spec")
|
||||
if lifetime == nil || configuration == nil {
|
||||
return ErrSupervision
|
||||
}
|
||||
defer configuration.Close()
|
||||
// PR_SET_CHILD_SUBREAPER is a Linux facility independent of the init system.
|
||||
if _, _, errno := syscall.Syscall6(syscall.SYS_PRCTL, 36, 1, 0, 0, 0, 0); errno != 0 {
|
||||
lifetime.Close()
|
||||
return fmt.Errorf("%w: subreaper: %w", ErrSupervision, errno)
|
||||
}
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
readerDone := make(chan struct{})
|
||||
go func() {
|
||||
defer close(readerDone)
|
||||
_, _ = io.Copy(io.Discard, lifetime)
|
||||
cancel()
|
||||
}()
|
||||
defer func() { lifetime.Close(); <-readerDone }()
|
||||
// Closing this descriptor also interrupts a partial startup on termination.
|
||||
stop := context.AfterFunc(ctx, func() { configuration.Close() })
|
||||
defer stop()
|
||||
data, err := io.ReadAll(io.LimitReader(configuration, MaxFrameBytes+1))
|
||||
if err != nil || len(data) > MaxFrameBytes {
|
||||
// No child exists yet. A partial configuration can result from startup
|
||||
// cancellation. Protocol EOF fails registration, while a zero helper
|
||||
// exit certifies cleanup and lets the host release its admission slot.
|
||||
return nil
|
||||
}
|
||||
configuration.Close()
|
||||
var spec Spec
|
||||
if decode_strict(data, &spec) != nil || validate_spec(spec) != nil {
|
||||
return nil
|
||||
}
|
||||
if ctx.Err() != nil {
|
||||
return nil
|
||||
}
|
||||
// Pdeathsig follows the creating thread; retain it for the child lifetime.
|
||||
runtime.LockOSThread()
|
||||
defer runtime.UnlockOSThread()
|
||||
command := exec.Command(spec.Program[0], spec.Program[1:]...)
|
||||
command.Dir = spec.Directory
|
||||
command.Stdin, command.Stdout, command.Stderr = os.Stdin, os.Stdout, os.Stderr
|
||||
command.SysProcAttr = &syscall.SysProcAttr{Setpgid: true, Pdeathsig: syscall.SIGKILL}
|
||||
if err := command.Start(); err != nil {
|
||||
// No process was started. Protocol EOF reports the startup failure;
|
||||
// successful cleanup keeps the manager available for a later retry.
|
||||
return nil
|
||||
}
|
||||
exited := make(chan struct{})
|
||||
go func() { _ = command.Wait(); close(exited) }()
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
_ = command.Process.Kill()
|
||||
<-exited
|
||||
case <-exited:
|
||||
}
|
||||
return reap_supervised_children(DefaultResourceLimits().CleanupTimeout)
|
||||
}
|
||||
|
||||
// Only this helper's children are killed. A child PID cannot be reused until
|
||||
// it has been reaped, and no other goroutine calls Wait after command.Wait joins.
|
||||
// Killing direct children adopts the next generation, including detached ones.
|
||||
func reap_supervised_children(timeout time.Duration) error {
|
||||
deadline := time.NewTimer(timeout)
|
||||
defer deadline.Stop()
|
||||
tick := time.NewTicker(5 * time.Millisecond)
|
||||
defer tick.Stop()
|
||||
for {
|
||||
paths, err := filepath.Glob("/proc/self/task/*/children")
|
||||
if err != nil || len(paths) == 0 {
|
||||
return ErrSupervision
|
||||
}
|
||||
for _, path := range paths {
|
||||
data, err := os.ReadFile(path)
|
||||
if os.IsNotExist(err) || errors.Is(err, syscall.ESRCH) { // A runtime thread exited during the scan.
|
||||
continue
|
||||
}
|
||||
if err != nil {
|
||||
return ErrSupervision
|
||||
}
|
||||
for _, value := range strings.Fields(string(data)) {
|
||||
pid, err := strconv.Atoi(value)
|
||||
if err != nil || pid <= 1 {
|
||||
return ErrSupervision
|
||||
}
|
||||
if err := syscall.Kill(pid, syscall.SIGKILL); err != nil && err != syscall.ESRCH {
|
||||
return fmt.Errorf("%w: terminate descendant: %w", ErrSupervision, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
for {
|
||||
var status syscall.WaitStatus
|
||||
pid, err := syscall.Wait4(-1, &status, syscall.WNOHANG, nil)
|
||||
if err == syscall.ECHILD {
|
||||
return nil
|
||||
}
|
||||
if err == syscall.EINTR {
|
||||
continue
|
||||
}
|
||||
if err != nil {
|
||||
return fmt.Errorf("%w: reap descendant: %w", ErrSupervision, err)
|
||||
}
|
||||
if pid == 0 {
|
||||
break
|
||||
}
|
||||
}
|
||||
select {
|
||||
case <-deadline.C:
|
||||
return fmt.Errorf("%w: descendant cleanup deadline expired", ErrSupervision)
|
||||
case <-tick.C:
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,210 @@
|
||||
// Copyright (c) 2026 Storm Dragon
|
||||
// SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
package tentacle
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"syscall"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func supervisor_program(t testing.TB) []string {
|
||||
t.Helper()
|
||||
executable, err := os.Executable()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return []string{executable, "-test.run=^TestSupervisorHelper$", "--", "supervisor"}
|
||||
}
|
||||
|
||||
func supervised_manager(t testing.TB) *Manager {
|
||||
t.Helper()
|
||||
manager, err := NewSupervisedManager(supervisor_program(t), "auto")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 7*time.Second)
|
||||
defer cancel()
|
||||
if err := manager.Shutdown(ctx); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
if manager.Stats() != (Stats{}) {
|
||||
t.Error(manager.Stats())
|
||||
}
|
||||
})
|
||||
return manager
|
||||
}
|
||||
|
||||
func supervised_descendant(t testing.TB, session *Session) int {
|
||||
t.Helper()
|
||||
result, err := session.Invoke(context.Background(), "echo", nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
value := wait_result(t, result)
|
||||
var pid int
|
||||
if value.Err != nil || json.Unmarshal(value.Value, &pid) != nil || pid <= 1 {
|
||||
t.Fatalf("descendant: %+v", value)
|
||||
}
|
||||
return pid
|
||||
}
|
||||
|
||||
func wait_supervised_reap(t *testing.T, pid int) {
|
||||
t.Helper()
|
||||
deadline := time.NewTimer(6 * time.Second)
|
||||
defer deadline.Stop()
|
||||
ticker := time.NewTicker(time.Millisecond)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
if _, err := os.Stat("/proc/" + strconv.Itoa(pid)); os.IsNotExist(err) || errors.Is(err, syscall.ESRCH) {
|
||||
return
|
||||
}
|
||||
select {
|
||||
case <-deadline.C:
|
||||
t.Fatalf("supervisor did not reap descendant %d", pid)
|
||||
case <-ticker.C:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestSupervisorHelper(t *testing.T) {
|
||||
if os.Args[len(os.Args)-1] == "supervisor" {
|
||||
if err := RunSupervisor(context.Background()); err != nil {
|
||||
os.Exit(1)
|
||||
}
|
||||
os.Exit(0)
|
||||
}
|
||||
if len(os.Args) < 3 || os.Args[len(os.Args)-2] != "supervisor-owner" {
|
||||
return
|
||||
}
|
||||
marker := os.Args[len(os.Args)-1]
|
||||
manager, err := NewSupervisedManager(supervisor_program(t), "auto")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
session := start_active(t, manager, "sample", "spawn-chain")
|
||||
pid := supervised_descendant(t, session)
|
||||
if err := os.WriteFile(marker, []byte(strconv.Itoa(pid)), 0600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// Abrupt reader death: no Disable, Shutdown, deferred cleanup or signals.
|
||||
os.Exit(0)
|
||||
}
|
||||
|
||||
func TestSupervisionWithoutInitToolsCleansDetachedDescendants(t *testing.T) {
|
||||
t.Setenv("PATH", "")
|
||||
t.Setenv("CTHULHU_TENTACLE_CGROUP_PARENT", "")
|
||||
for _, mode := range []string{"spawn", "spawn-detached", "spawn-chain"} {
|
||||
t.Run(mode, func(t *testing.T) {
|
||||
manager := supervised_manager(t)
|
||||
first := start_active(t, manager, "first", mode)
|
||||
second := start_active(t, manager, "second", "spawn-detached")
|
||||
pid := supervised_descendant(t, first)
|
||||
other := supervised_descendant(t, second)
|
||||
first.Disable()
|
||||
wait_signal(t, first.Done())
|
||||
if state, err := first.Status(); state != Stopped || err != nil {
|
||||
t.Fatalf("disable: %s %v", state, err)
|
||||
}
|
||||
wait_supervised_reap(t, pid)
|
||||
if _, err := os.Stat("/proc/" + strconv.Itoa(other)); err != nil {
|
||||
t.Fatal("another session was terminated", err)
|
||||
}
|
||||
second.Disable()
|
||||
wait_signal(t, second.Done())
|
||||
wait_supervised_reap(t, other)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestSupervisionCleansDescendantsAfterReaderDeath(t *testing.T) {
|
||||
t.Setenv("PATH", "")
|
||||
t.Setenv("CTHULHU_TENTACLE_CGROUP_PARENT", "")
|
||||
executable, _ := os.Executable()
|
||||
marker := filepath.Join(t.TempDir(), "descendant")
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
|
||||
defer cancel()
|
||||
command := exec.CommandContext(ctx, executable, "-test.run=^TestSupervisorHelper$", "--", "supervisor-owner", marker)
|
||||
if output, err := command.CombinedOutput(); err != nil {
|
||||
t.Fatalf("owner: %v %s", err, output)
|
||||
}
|
||||
data, err := os.ReadFile(marker)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
pid, err := strconv.Atoi(string(data))
|
||||
if err != nil || pid <= 1 {
|
||||
t.Fatal("invalid descendant marker")
|
||||
}
|
||||
wait_supervised_reap(t, pid)
|
||||
}
|
||||
|
||||
func TestSupervisionMissingProgramReleasesCapacity(t *testing.T) {
|
||||
t.Setenv("CTHULHU_TENTACLE_CGROUP_PARENT", "")
|
||||
manager := supervised_manager(t)
|
||||
spec := Spec{ID: "sample", Program: []string{filepath.Join(t.TempDir(), "missing")}}
|
||||
session, err := manager.Start(context.Background(), spec)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
wait_signal(t, session.Done())
|
||||
if state, failure := session.Status(); state != Failed || failure == nil {
|
||||
t.Fatal(state, failure)
|
||||
}
|
||||
healthy := start_active(t, manager, "sample", "echo")
|
||||
healthy.Disable()
|
||||
wait_signal(t, healthy.Done())
|
||||
}
|
||||
|
||||
func TestSupervisionRejectsInvalidExplicitCgroupParent(t *testing.T) {
|
||||
for _, parent := range []string{"relative", t.TempDir()} {
|
||||
if _, err := NewSupervisedManager(supervisor_program(t), parent); !errors.Is(err, ErrContainment) {
|
||||
t.Fatalf("silently ignored explicit parent %q: %v", parent, err)
|
||||
}
|
||||
}
|
||||
t.Setenv("CTHULHU_TENTACLE_CGROUP_PARENT", t.TempDir())
|
||||
if _, err := NewSupervisedManager(supervisor_program(t), "auto"); !errors.Is(err, ErrContainment) {
|
||||
t.Fatalf("silently ignored configured parent: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSupervisionCancelledStartupReleasesCapacity(t *testing.T) {
|
||||
t.Setenv("CTHULHU_TENTACLE_CGROUP_PARENT", "")
|
||||
manager := supervised_manager(t)
|
||||
session, err := manager.Start(context.Background(), child_spec(t, "sample", "hang-start"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
wait_control(t, func() bool {
|
||||
session.mutex.Lock()
|
||||
defer session.mutex.Unlock()
|
||||
return len(session.pending) > 0
|
||||
})
|
||||
session.Disable()
|
||||
wait_signal(t, session.Done())
|
||||
if state, err := session.Status(); state != Stopped || err != nil {
|
||||
t.Fatal(state, err)
|
||||
}
|
||||
healthy := start_active(t, manager, "sample", "echo")
|
||||
healthy.Disable()
|
||||
wait_signal(t, healthy.Done())
|
||||
}
|
||||
|
||||
func TestSupervisionDetectsChildExitAfterActivation(t *testing.T) {
|
||||
t.Setenv("CTHULHU_TENTACLE_CGROUP_PARENT", "")
|
||||
manager := supervised_manager(t)
|
||||
session := start_active(t, manager, "sample", "exit-active")
|
||||
wait_signal(t, session.Done())
|
||||
if state, err := session.Status(); state != Failed || !errors.Is(err, ErrExited) {
|
||||
t.Fatal(state, err)
|
||||
}
|
||||
}
|
||||
@@ -1,5 +1,5 @@
|
||||
#!/usr/bin/env bash
|
||||
# Explicit local development launcher; requires systemd 254+ and cgroup v2.
|
||||
# Optional cgroup development launcher, independent of any init system.
|
||||
set -euo pipefail
|
||||
if (( $# == 0 )); then
|
||||
echo 'Usage: bash tests/run-delegated-reader.sh /absolute/path/to/cthulhu [reader options]' >&2
|
||||
@@ -9,18 +9,10 @@ if [[ $1 != /* || ! -x $1 ]]; then
|
||||
echo 'The reader executable must be an absolute executable path.' >&2
|
||||
exit 2
|
||||
fi
|
||||
# Preserve desktop/backend locations without importing the whole environment
|
||||
# into the persistent user manager. These values apply only to this service.
|
||||
environment=()
|
||||
for name in DISPLAY XAUTHORITY DBUS_SESSION_BUS_ADDRESS AT_SPI_BUS_ADDRESS \
|
||||
XDG_RUNTIME_DIR XDG_SESSION_TYPE XDG_SESSION_DESKTOP XDG_DATA_HOME \
|
||||
XDG_CONFIG_HOME XDG_CACHE_HOME LANG LANGUAGE LC_ALL \
|
||||
SPEECHD_ADDRESS SPEECHD_SOCKET; do
|
||||
if [[ -v $name ]]; then
|
||||
environment+=("--setenv=$name=${!name}")
|
||||
parent=${CTHULHU_TENTACLE_CGROUP_PARENT:-}
|
||||
if [[ $parent != /* || ! -d $parent || ! -w $parent ]]; then
|
||||
echo 'Set CTHULHU_TENTACLE_CGROUP_PARENT to an already delegated, writable cgroup v2 parent.' >&2
|
||||
echo 'Ordinary Tentacle use needs no launcher or cgroup setup.' >&2
|
||||
exit 2
|
||||
fi
|
||||
done
|
||||
exec systemd-run --user --wait --pipe --collect --same-dir \
|
||||
--property='Delegate=cpu memory pids' --property=DelegateSubgroup=reader \
|
||||
--property=KillMode=control-group --property=TimeoutStopSec=10s \
|
||||
"${environment[@]}" -- "$@"
|
||||
exec "$@" --tentacle-cgroup-parent "$parent"
|
||||
Reference in new issue
Block a user