Code review and cleanup.

This commit is contained in:
ak47 committed 2026-11-01 22:01:32 -05:00
1 parent bbe931335f
commit 92d6ad3cc4
23 files changed
+1497 -886

No files matched your search

+268
View File
@@ -0,0 +1,268 @@
// Copyright (c) 2026 Storm Dragon
// Authors: Storm Dragon, Angus Kola
// SPDX-License-Identifier: GPL-3.0-or-later
package app
import (
"context"
"errors"
"slices"
"time"
"git.stormux.org/storm/cthulhu/internal/tentacle"
)
// 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
previous *tentacle.Entry
}
// A retired package stays on disk until the session using it has joined.
type tentacleRetiredPackage struct {
entry tentacle.Entry
session *tentacle.Session
}
// tentacleLifecycle owns managed state for one run, on the lifecycle goroutine.
// Watchers and cancellation callbacks only notify it; they never mutate these
// maps or transitions. Startup rollback runs before shutdown joins sessions.
type tentacleLifecycle struct {
ctx context.Context
runtime *tentacleRuntime
entries []tentacle.Entry
repositories []tentacle.Repository
loadError error
pending map[string]bool
sessions map[string]*tentacle.Session
problems map[string]string
startups map[string]tentacleStartupCall
removedPackages map[string]tentacleRetiredPackage
wake chan struct{}
manager *tentacle.Manager
current *tentacle.Session
ready <-chan struct{}
active []*tentacle.Session
watchers []<-chan struct{}
}
func (runtime *tentacleRuntime) run_managed(ctx context.Context) error {
entries, loadError := runtime.registry.Load()
owner := &tentacleLifecycle{
ctx: ctx,
runtime: runtime,
entries: entries,
loadError: loadError,
repositories: runtime.registry.Repositories(),
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]tentacleRetiredPackage),
wake: make(chan struct{}, 1),
}
if loadError == nil {
if err := tentacle.RecoverPackages(tentacle.PackageRoot(runtime.registry.Path), entries); err != nil {
runtime.logger.Warn("recover tentacle packages", "error", err)
}
}
for _, entry := range entries {
owner.pending[entry.ID] = entry.Enabled
}
owner.setSnapshot()
if loadError != nil {
runtime.report("Tentacles could not start.")
}
defer owner.shutdown()
defer owner.cancelStartups()
return owner.run()
}
func (owner *tentacleLifecycle) notify() {
select {
case owner.wake <- struct{}{}:
default:
}
}
func (owner *tentacleLifecycle) setSnapshot() {
result := tentacle.Snapshot{Repositories: tentacle.CloneRepositories(owner.repositories)}
if owner.loadError != nil {
result.Message = "Tentacle registry could not be loaded; no tentacles were started."
}
for _, entry := range owner.entries {
status := tentacle.EntryStatus{Entry: entry, State: "disabled", Message: owner.problems[entry.ID]}
if entry.Enabled {
status.State = "starting"
}
if session := owner.sessions[entry.ID]; session != nil {
state, _ := session.Status()
status.State = string(state)
if state == tentacle.Ready {
status.State = "starting"
}
if (state == tentacle.Failed || state == tentacle.Stopped) && entry.Enabled && status.Message == "" {
status.Message = "Tentacle stopped; retry to start it again."
}
}
if owner.pending[entry.ID] {
status.State = "starting"
}
if status.Message != "" && !owner.pending[entry.ID] {
status.State = "failed"
}
if !entry.Enabled && owner.sessions[entry.ID] == nil {
status.State = "disabled"
}
result.Entries = append(result.Entries, status)
}
result.Updates = tentacle.AvailableUpdates(result)
owner.runtime.mutex.Lock()
owner.runtime.snapshot = result
owner.runtime.mutex.Unlock()
}
func (owner *tentacleLifecycle) shutdown() {
owner.runtime.Disable()
cleanup, cancel := context.WithTimeout(context.Background(), 7*time.Second)
defer cancel()
var bindingError error
if owner.runtime.input != nil {
bindingError = owner.runtime.input.SetTentacleBindings(cleanup, nil)
}
if owner.runtime.readerContext.Err() != nil {
bindingError = nil
}
var managerError error
if owner.manager != nil {
managerError = owner.manager.Shutdown(cleanup)
}
for _, retired := range owner.removedPackages {
entry := retired.entry
if managerError == nil {
if err := tentacle.RemovePackage(tentacle.PackageRoot(owner.runtime.registry.Path), entry); err != nil {
owner.runtime.logger.Warn("remove tentacle package", "error", err)
}
}
}
for _, done := range owner.watchers {
<-done
}
owner.runtime.results.Wait()
owner.runtime.stopError = errors.Join(bindingError, managerError)
if owner.runtime.stopError != nil && owner.runtime.readerContext.Err() == nil {
select {
case owner.runtime.failures <- owner.runtime.stopError:
default:
}
}
}
func (owner *tentacleLifecycle) reconcileBindings() error {
retained := owner.active[:0]
for _, session := range owner.active {
if state, _ := session.Status(); state == tentacle.Active {
retained = append(retained, session)
}
}
changed := len(retained) != len(owner.active)
owner.active = retained
if changed {
owner.runtime.publishCommands(owner.active)
if err := owner.runtime.bind(owner.ctx, owner.active); err != nil && owner.ctx.Err() == nil {
select {
case owner.runtime.failures <- err:
default:
}
return err
}
}
return nil
}
func (owner *tentacleLifecycle) reapSessions() {
for id, session := range owner.sessions {
select {
case <-session.Done():
if state, failure := session.Status(); state == tentacle.Failed {
owner.runtime.logger.Warn("configured tentacle failed", "tentacle", id, "error", failure)
}
if !owner.pending[id] {
index := slices.IndexFunc(owner.entries, func(entry tentacle.Entry) bool { return entry.ID == id })
if index >= 0 && owner.entries[index].Enabled && owner.problems[id] == "" {
owner.problems[id] = "Tentacle stopped; retry to start it again."
owner.runtime.report("A tentacle stopped unexpectedly.")
}
}
delete(owner.sessions, id)
owner.runtime.mutex.Lock()
owner.runtime.sessions = slices.DeleteFunc(owner.runtime.sessions, func(value *tentacle.Session) bool { return value == session })
owner.runtime.mutex.Unlock()
default:
}
}
}
func (owner *tentacleLifecycle) removeRetiredPackages() {
for directory, retired := range owner.removedPackages {
entry := retired.entry
if retired.session != nil {
select {
case <-retired.session.Done():
default:
continue
}
}
if slices.ContainsFunc(owner.entries, func(candidateEntry tentacle.Entry) bool {
return candidateEntry.PackageDirectory == directory
}) {
continue
}
if err := tentacle.RemovePackage(tentacle.PackageRoot(owner.runtime.registry.Path), entry); err != nil {
owner.runtime.logger.Warn("remove tentacle package", "error", err)
owner.runtime.report("Tentacle was removed, but its package files could not be deleted.")
continue
}
delete(owner.removedPackages, directory)
}
}
func (owner *tentacleLifecycle) run() error {
for {
if owner.ctx.Err() != nil {
return nil
}
for id, startup := range owner.startups {
if startup.call.ctx.Err() != nil {
owner.failStartup(id, "Tentacle startup was cancelled before activation.")
}
}
// Reconcile revoked sessions before staging new native grabs, and release
// joined sessions before starting replacements or removing their files.
if err := owner.reconcileBindings(); err != nil {
return err
}
owner.reapSessions()
if owner.current == nil {
owner.removeRetiredPackages()
owner.startPending()
}
owner.setSnapshot()
select {
case <-owner.ctx.Done():
return nil
case <-owner.wake:
case call := <-owner.runtime.requests:
owner.handleRequest(call)
case <-owner.ready:
if err := owner.activateReady(); err != nil {
return err
}
}
}
}
+124
View File
@@ -0,0 +1,124 @@
// Copyright (c) 2026 Storm Dragon
// Authors: Storm Dragon, Angus Kola
// SPDX-License-Identifier: GPL-3.0-or-later
package app
import (
"context"
"errors"
"reflect"
"slices"
"strings"
"git.stormux.org/storm/cthulhu/internal/tentacle"
)
func (owner *tentacleLifecycle) handleRequest(call tentacleManagementCall) {
if err := call.ctx.Err(); err != nil {
call.result <- err
return
}
if owner.loadError != nil {
call.result <- errors.New("tentacle registry could not be loaded; repair it and restart")
return
}
owner.runtime.mutex.Lock()
disabled := owner.runtime.disabled
owner.runtime.mutex.Unlock()
if disabled {
call.result <- tentacle.ErrDisabled
return
}
if strings.HasPrefix(call.request.Action, "repository-") {
next, err := change_tentacle_repositories(owner.repositories, call.request, call.repository)
if err == nil {
err = owner.runtime.registry.SaveState(owner.entries, next)
}
if err == nil || errors.Is(err, tentacle.ErrRegistrySync) {
owner.repositories = next
owner.setSnapshot()
}
call.result <- err
return
}
if call.packageFiles != nil {
index := slices.IndexFunc(owner.repositories, func(value tentacle.Repository) bool { return value.URL == call.repository.URL })
if index < 0 || !reflect.DeepEqual(owner.repositories[index], *call.repository) {
call.result <- errors.New("Repository changed during installation; refresh and try again.")
return
}
}
if _, starting := owner.startups[call.request.ID]; starting && call.request.Action != "disable" && call.request.Action != "remove" {
call.result <- tentacle.ErrBusy
return
}
if startup, exists := owner.startups[call.request.ID]; exists && startup.previous != nil {
owner.failStartup(call.request.ID, "Tentacle update was cancelled by another operation.")
}
next, err := change_tentacle_entries(owner.entries, call.request)
if err == nil && call.request.Action != "retry" && reflect.DeepEqual(next, owner.entries) {
// Saved enablement is intent, not proof of a live session. An
// explicit enable/update must recover a failed or starting entry.
index := slices.IndexFunc(next, func(entry tentacle.Entry) bool { return entry.ID == call.request.ID })
state := tentacle.Stopped
if session := owner.sessions[call.request.ID]; session != nil {
state, _ = session.Status()
}
if !next[index].Enabled || state == tentacle.Active {
call.result <- nil
return
}
}
if err == nil && call.request.Action != "retry" {
err = owner.runtime.registry.Save(next)
}
if err == nil || errors.Is(err, tentacle.ErrRegistrySync) {
id := call.request.ID
var previous *tentacle.Entry
if call.packageFiles != nil && call.request.Action == "update" {
index := slices.IndexFunc(owner.entries, func(entry tentacle.Entry) bool { return entry.ID == id })
if index >= 0 && owner.entries[index].Enabled {
saved := tentacle.CloneEntries(owner.entries[index : index+1])[0]
previous = &saved
}
}
owner.replyStartup(id, errors.New("Tentacle startup was cancelled by another operation."))
if call.packageFiles != nil {
call.packageFiles.Retain()
}
if call.request.Action == "remove" || call.request.Action == "update" {
for _, entry := range owner.entries {
if entry.ID != id || entry.PackageDirectory == "" {
continue
}
if call.request.Action == "update" && (previous != nil ||
entry.PackageDirectory == call.request.Entry.PackageDirectory) {
continue
}
owner.removedPackages[entry.PackageDirectory] = tentacleRetiredPackage{entry: entry, session: owner.sessions[id]}
}
}
owner.revoke(id)
if owner.current != nil && owner.current.TentacleID() == id {
owner.current, owner.ready = nil, nil
}
owner.entries = next
delete(owner.problems, id)
delete(owner.pending, id)
for _, entry := range owner.entries {
if entry.ID == id {
owner.pending[id] = entry.Enabled
}
}
// Immediately withdraw dispatch, without waiting for key release.
stillActive := slices.DeleteFunc(slices.Clone(owner.active), func(session *tentacle.Session) bool { state, _ := session.Status(); return state != tentacle.Active })
owner.runtime.publishCommands(stillActive)
owner.setSnapshot()
if owner.pending[id] {
owner.startups[id] = tentacleStartupCall{call: call, saveError: err, stop: context.AfterFunc(call.ctx, owner.notify), previous: previous}
return
}
}
call.result <- err
}
+9 -561
View File
@@ -7,10 +7,7 @@ import (
"context"
"errors"
"fmt"
"reflect"
"slices"
"strings"
"time"
"git.stormux.org/storm/cthulhu/internal/input"
"git.stormux.org/storm/cthulhu/internal/tentacle"
@@ -24,21 +21,6 @@ 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
previous *tentacle.Entry
}
// A retired package stays on disk until the session using it has joined.
type tentacleRetiredPackage struct {
entry tentacle.Entry
session *tentacle.Session
}
func (runtime *tentacleRuntime) Snapshot() tentacle.Snapshot {
if runtime == nil {
return tentacle.Snapshot{Suppressed: true, Message: "Tentacle management is unavailable for this run."}
@@ -133,67 +115,15 @@ func (runtime *tentacleRuntime) Manage(ctx context.Context, request tentacle.Man
return tentacle.ErrDisabled
}
if request.Action == "install" || request.Action == "install-enable" || request.Action == "upgrade" {
if request.Entry != nil {
return errors.New("unexpected tentacle entry in installation request")
repository, err := repositoryForInstallation(runtime.Snapshot(), request)
if err != nil {
return err
}
snapshot := runtime.Snapshot()
if snapshot.Suppressed || snapshot.Message != "" {
return tentacle.ErrUnavailable
}
index := slices.IndexFunc(snapshot.Repositories, func(value tentacle.Repository) bool { return value.URL == request.URL })
if index < 0 {
return errors.New("repository is not configured")
}
repository := snapshot.Repositories[index]
id := tentacle.PackageID(repository, request.ID)
installed := slices.ContainsFunc(snapshot.Entries, func(value tentacle.EntryStatus) bool { return value.Entry.ID == id })
if request.Action == "upgrade" && !installed {
return errors.New("tentacle is not installed")
}
if request.Action != "upgrade" && installed {
return errors.New("tentacle is already installed")
}
if request.Action == "upgrade" {
for _, status := range snapshot.Entries {
if status.Entry.ID != id {
continue
}
for _, manifest := range repository.Tentacles {
if manifest.ID != request.ID {
continue
}
if manifest.Version == status.Entry.Version {
return errors.New("tentacle is already at the repository version")
}
newer, err := tentacle.IsNewerVersion(manifest.Version, status.Entry.Version)
if err != nil {
return fmt.Errorf("cannot compare tentacle versions: %w", err)
}
if !newer {
return errors.New("repository tentacle version is not newer than installed version")
}
break
}
break
}
}
runtime.mutex.Lock()
if runtime.repositoryBusy {
runtime.mutex.Unlock()
if !runtime.beginRepositoryWork() {
return tentacle.ErrBusy
}
runtime.repositoryBusy = true
runtime.mutex.Unlock()
defer func() { runtime.mutex.Lock(); runtime.repositoryBusy = false; runtime.mutex.Unlock() }()
prepare := runtime.preparePackage
if prepare == nil {
prepare = tentacle.PreparePackage
}
prepareCtx, cancel := context.WithTimeout(ctx, 45*time.Second)
stop := context.AfterFunc(runtime.lifecycleContext, cancel)
prepared, err := prepare(prepareCtx, tentacle.PackageRoot(runtime.registry.Path), repository, request.ID)
stop()
cancel()
defer runtime.endRepositoryWork()
prepared, err := runtime.prepareManagedPackage(ctx, repository, request.ID)
if err != nil {
return err
}
@@ -216,26 +146,12 @@ func (runtime *tentacleRuntime) Manage(ctx context.Context, request tentacle.Man
if err != nil {
return err
}
runtime.mutex.Lock()
if runtime.repositoryBusy {
runtime.mutex.Unlock()
if !runtime.beginRepositoryWork() {
return tentacle.ErrBusy
}
runtime.repositoryBusy = true
runtime.mutex.Unlock()
defer func() { runtime.mutex.Lock(); runtime.repositoryBusy = false; runtime.mutex.Unlock() }()
defer runtime.endRepositoryWork()
request.URL, call.request.URL = origin, origin
fetch := runtime.fetchRepository
if fetch == nil {
fetch = tentacle.FetchRepository
}
// Fetch on the Preferences worker, leaving lifecycle and native input
// free to disable, revoke, and process failures while the network waits.
fetchCtx, cancel := context.WithTimeout(ctx, 45*time.Second)
stop := context.AfterFunc(runtime.lifecycleContext, cancel)
repository, err := fetch(fetchCtx, origin)
stop()
cancel()
repository, err := runtime.fetchManagedRepository(ctx, origin)
if err != nil {
return err
}
@@ -298,471 +214,3 @@ func change_tentacle_entries(entries []tentacle.Entry, request tentacle.Manageme
}
return next, tentacle.ValidateEntries(next)
}
func (runtime *tentacleRuntime) run_managed(ctx context.Context) error {
entries, loadError := runtime.registry.Load()
repositories := runtime.registry.Repositories()
if loadError == nil {
if err := tentacle.RecoverPackages(tentacle.PackageRoot(runtime.registry.Path), entries); err != nil {
runtime.logger.Warn("recover tentacle packages", "error", err)
}
}
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]tentacleRetiredPackage)
for _, entry := range entries {
pending[entry.ID] = entry.Enabled
}
wake := make(chan struct{}, 1)
notify := func() {
select {
case wake <- struct{}{}:
default:
}
}
var manager *tentacle.Manager
var current *tentacle.Session
var ready <-chan struct{}
var active []*tentacle.Session
var watchers []<-chan struct{}
setSnapshot := func() {
result := tentacle.Snapshot{Repositories: tentacle.CloneRepositories(repositories)}
if loadError != nil {
result.Message = "Tentacle registry could not be loaded; no tentacles were started."
}
for _, entry := range entries {
status := tentacle.EntryStatus{Entry: entry, State: "disabled", Message: problems[entry.ID]}
if entry.Enabled {
status.State = "starting"
}
if session := sessions[entry.ID]; session != nil {
state, _ := session.Status()
status.State = string(state)
if state == tentacle.Ready {
status.State = "starting"
}
if (state == tentacle.Failed || state == tentacle.Stopped) && entry.Enabled && status.Message == "" {
status.Message = "Tentacle stopped; retry to start it again."
}
}
if pending[entry.ID] {
status.State = "starting"
}
if status.Message != "" && !pending[entry.ID] {
status.State = "failed"
}
if !entry.Enabled && sessions[entry.ID] == nil {
status.State = "disabled"
}
result.Entries = append(result.Entries, status)
}
result.Updates = tentacle.AvailableUpdates(result)
runtime.mutex.Lock()
runtime.snapshot = result
runtime.mutex.Unlock()
}
setSnapshot()
if loadError != nil {
runtime.report("Tentacles could not start.")
}
defer func() {
runtime.Disable()
cleanup, cancel := context.WithTimeout(context.Background(), 7*time.Second)
defer cancel()
var bindingError error
if runtime.input != nil {
bindingError = runtime.input.SetTentacleBindings(cleanup, nil)
}
if runtime.readerContext.Err() != nil {
bindingError = nil
}
var managerError error
if manager != nil {
managerError = manager.Shutdown(cleanup)
}
for _, retired := range removedPackages {
entry := retired.entry
if managerError == nil {
if err := tentacle.RemovePackage(tentacle.PackageRoot(runtime.registry.Path), entry); err != nil {
runtime.logger.Warn("remove tentacle package", "error", err)
}
}
}
for _, done := range watchers {
<-done
}
runtime.results.Wait()
runtime.stopError = errors.Join(bindingError, managerError)
if runtime.stopError != nil && runtime.readerContext.Err() == nil {
select {
case runtime.failures <- runtime.stopError:
default:
}
}
}()
ensureManager := func() error {
if manager != nil {
return nil
}
if runtime.input == nil {
return errors.New("native bindings are unavailable")
}
var err error
manager, err = runtime.makeManager()
return err
}
revoke := func(id string) {
if session := sessions[id]; session != nil {
session.Disable()
runtime.output.RetireRevoked()
}
}
replyStartup := func(id string, err error) {
startup, exists := startups[id]
if !exists {
return
}
delete(startups, id)
startup.stop()
if err == nil {
if startup.previous != nil {
previous := *startup.previous
// Starting the replacement already waited for the old session.
removedPackages[previous.PackageDirectory] = tentacleRetiredPackage{entry: previous}
}
err = startup.saveError
}
setSnapshot()
startup.call.result <- err
}
failStartup := func(id, message string) {
problems[id] = message
startup, exists := startups[id]
if !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 {
if startup.previous != nil {
next[index] = *startup.previous
} else {
next[index].Enabled = false
}
}
}
err := runtime.registry.Save(next)
if err == nil || errors.Is(err, tentacle.ErrRegistrySync) {
if startup.previous != nil {
failed := slices.IndexFunc(entries, func(entry tentacle.Entry) bool { return entry.ID == id })
if failed >= 0 {
entry := entries[failed]
removedPackages[entry.PackageDirectory] = tentacleRetiredPackage{entry: entry, session: sessions[id]}
}
runtime.mutex.Lock()
disabled := runtime.disabled
runtime.mutex.Unlock()
pending[id] = startup.previous.Enabled && !disabled && ctx.Err() == nil
message += " The previous package was restored."
}
entries = next
}
if err != nil {
message += " The recovery 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 {
if state, _ := session.Status(); state == tentacle.Active {
retained = append(retained, session)
}
}
changed := len(retained) != len(active)
active = retained
if changed {
runtime.publishCommands(active)
if err := runtime.bind(ctx, active); err != nil && ctx.Err() == nil {
select {
case runtime.failures <- err:
default:
}
return err
}
}
// Joined sessions release host capacity before an explicit retry/update.
for id, session := range sessions {
select {
case <-session.Done():
if state, failure := session.Status(); state == tentacle.Failed {
runtime.logger.Warn("configured tentacle failed", "tentacle", id, "error", failure)
}
if !pending[id] {
index := slices.IndexFunc(entries, func(entry tentacle.Entry) bool { return entry.ID == id })
if index >= 0 && entries[index].Enabled && problems[id] == "" {
problems[id] = "Tentacle stopped; retry to start it again."
runtime.report("A tentacle stopped unexpectedly.")
}
}
delete(sessions, id)
runtime.mutex.Lock()
runtime.sessions = slices.DeleteFunc(runtime.sessions, func(value *tentacle.Session) bool { return value == session })
runtime.mutex.Unlock()
default:
}
}
if current == nil {
for directory, retired := range removedPackages {
entry := retired.entry
if retired.session != nil {
select {
case <-retired.session.Done():
default:
continue
}
}
if slices.ContainsFunc(entries, func(active tentacle.Entry) bool {
return active.PackageDirectory == directory
}) {
continue
}
if err := tentacle.RemovePackage(tentacle.PackageRoot(runtime.registry.Path), entry); err != nil {
runtime.logger.Warn("remove tentacle package", "error", err)
runtime.report("Tentacle was removed, but its package files could not be deleted.")
continue
}
delete(removedPackages, directory)
}
for _, entry := range entries {
if !pending[entry.ID] || sessions[entry.ID] != nil {
continue
}
pending[entry.ID] = false
err := ensureManager()
var session *tentacle.Session
if err == nil {
session, err = manager.Start(ctx, entry.Spec())
}
if err != nil {
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()
break
}
sessions[entry.ID] = session
runtime.mutex.Lock()
runtime.sessions = append(runtime.sessions, session)
if runtime.disabled {
session.Disable()
}
runtime.mutex.Unlock()
current, ready = session, session.Ready()
done := make(chan struct{})
// Prune joined watcher handles so retries never accumulate history.
watchers = slices.DeleteFunc(watchers, func(done <-chan struct{}) bool {
select {
case <-done:
return true
default:
return false
}
})
watchers = append(watchers, done)
go func() {
defer close(done)
select {
case <-session.Revoked():
case <-ctx.Done():
}
runtime.output.RetireRevoked()
runtime.signal_extensions()
notify()
<-session.Done()
notify()
}()
break
}
}
setSnapshot()
select {
case <-ctx.Done():
return nil
case <-wake:
case call := <-runtime.requests:
if err := call.ctx.Err(); err != nil {
call.result <- err
continue
}
if loadError != nil {
call.result <- errors.New("tentacle registry could not be loaded; repair it and restart")
continue
}
runtime.mutex.Lock()
disabled := runtime.disabled
runtime.mutex.Unlock()
if disabled {
call.result <- tentacle.ErrDisabled
continue
}
if strings.HasPrefix(call.request.Action, "repository-") {
next, err := change_tentacle_repositories(repositories, call.request, call.repository)
if err == nil {
err = runtime.registry.SaveState(entries, next)
}
if err == nil || errors.Is(err, tentacle.ErrRegistrySync) {
repositories = next
setSnapshot()
}
call.result <- err
continue
}
if call.packageFiles != nil {
index := slices.IndexFunc(repositories, func(value tentacle.Repository) bool { return value.URL == call.repository.URL })
if index < 0 || !reflect.DeepEqual(repositories[index], *call.repository) {
call.result <- errors.New("Repository changed during installation; refresh and try again.")
continue
}
}
if _, starting := startups[call.request.ID]; starting && call.request.Action != "disable" && call.request.Action != "remove" {
call.result <- tentacle.ErrBusy
continue
}
if startup, exists := startups[call.request.ID]; exists && startup.previous != nil {
failStartup(call.request.ID, "Tentacle update was cancelled by another operation.")
}
next, err := change_tentacle_entries(entries, call.request)
if err == nil && call.request.Action != "retry" && reflect.DeepEqual(next, entries) {
// Saved enablement is intent, not proof of a live session. An
// explicit enable/update must recover a failed or starting entry.
index := slices.IndexFunc(next, func(entry tentacle.Entry) bool { return entry.ID == call.request.ID })
state := tentacle.Stopped
if session := sessions[call.request.ID]; session != nil {
state, _ = session.Status()
}
if !next[index].Enabled || state == tentacle.Active {
call.result <- nil
continue
}
}
if err == nil && call.request.Action != "retry" {
err = runtime.registry.Save(next)
}
if err == nil || errors.Is(err, tentacle.ErrRegistrySync) {
id := call.request.ID
var previous *tentacle.Entry
if call.packageFiles != nil && call.request.Action == "update" {
index := slices.IndexFunc(entries, func(entry tentacle.Entry) bool { return entry.ID == id })
if index >= 0 && entries[index].Enabled {
saved := tentacle.CloneEntries(entries[index : index+1])[0]
previous = &saved
}
}
replyStartup(id, errors.New("Tentacle startup was cancelled by another operation."))
if call.packageFiles != nil {
call.packageFiles.Retain()
}
if call.request.Action == "remove" || call.request.Action == "update" {
for _, entry := range entries {
if entry.ID != id || entry.PackageDirectory == "" {
continue
}
if call.request.Action == "update" && (previous != nil ||
entry.PackageDirectory == call.request.Entry.PackageDirectory) {
continue
}
removedPackages[entry.PackageDirectory] = tentacleRetiredPackage{entry: entry, session: sessions[id]}
}
}
revoke(id)
if current != nil && current.TentacleID() == id {
current, ready = nil, nil
}
entries = next
delete(problems, id)
delete(pending, id)
for _, entry := range entries {
if entry.ID == id {
pending[id] = entry.Enabled
}
}
// Immediately withdraw dispatch, without waiting for key release.
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), previous: previous}
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)
if err := runtime.bind(ctx, candidate); err != nil {
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
}
// Native binding setup can wait for held keys. Cancellation during
// that wait must roll back the new grabs before reporting failure.
activationError := ctx.Err()
if startup, exists := startups[session.TentacleID()]; exists && activationError == nil {
activationError = startup.call.ctx.Err()
}
if activationError == nil {
activationError = session.Activate()
}
if activationError != nil {
session.Disable()
if err := runtime.bind(ctx, active); err != nil && ctx.Err() == nil {
select {
case runtime.failures <- err:
default:
}
return err
}
failStartup(session.TentacleID(), "Tentacle stopped before activation; enable it again to retry.")
continue
}
active = candidate
delete(problems, session.TentacleID())
runtime.publishCommands(active)
replyStartup(session.TentacleID(), nil)
runtime.logger.Info("configured tentacle active", "tentacle", session.TentacleID(), "session", session.Identity())
}
}
}
+108
View File
@@ -0,0 +1,108 @@
// Copyright (c) 2026 Storm Dragon
// Authors: Storm Dragon, Angus Kola
// SPDX-License-Identifier: GPL-3.0-or-later
package app
import (
"context"
"errors"
"fmt"
"slices"
"time"
"git.stormux.org/storm/cthulhu/internal/tentacle"
)
// Preparation runs on the caller's worker. Its busy reservation remains owned
// by Manage through the lifecycle reply, including rollback after cancellation.
func (runtime *tentacleRuntime) beginRepositoryWork() bool {
runtime.mutex.Lock()
defer runtime.mutex.Unlock()
if runtime.repositoryBusy {
return false
}
runtime.repositoryBusy = true
return true
}
func (runtime *tentacleRuntime) endRepositoryWork() {
runtime.mutex.Lock()
runtime.repositoryBusy = false
runtime.mutex.Unlock()
}
func repositoryForInstallation(snapshot tentacle.Snapshot, request tentacle.ManagementRequest) (tentacle.Repository, error) {
if request.Entry != nil {
return tentacle.Repository{}, errors.New("unexpected tentacle entry in installation request")
}
if snapshot.Suppressed || snapshot.Message != "" {
return tentacle.Repository{}, tentacle.ErrUnavailable
}
index := slices.IndexFunc(snapshot.Repositories, func(value tentacle.Repository) bool { return value.URL == request.URL })
if index < 0 {
return tentacle.Repository{}, errors.New("repository is not configured")
}
repository := snapshot.Repositories[index]
id := tentacle.PackageID(repository, request.ID)
installed := slices.ContainsFunc(snapshot.Entries, func(value tentacle.EntryStatus) bool { return value.Entry.ID == id })
if request.Action == "upgrade" && !installed {
return tentacle.Repository{}, errors.New("tentacle is not installed")
}
if request.Action != "upgrade" && installed {
return tentacle.Repository{}, errors.New("tentacle is already installed")
}
if request.Action == "upgrade" {
for _, status := range snapshot.Entries {
if status.Entry.ID != id {
continue
}
for _, manifest := range repository.Tentacles {
if manifest.ID != request.ID {
continue
}
if manifest.Version == status.Entry.Version {
return tentacle.Repository{}, errors.New("tentacle is already at the repository version")
}
newer, err := tentacle.IsNewerVersion(manifest.Version, status.Entry.Version)
if err != nil {
return tentacle.Repository{}, fmt.Errorf("cannot compare tentacle versions: %w", err)
}
if !newer {
return tentacle.Repository{}, errors.New("repository tentacle version is not newer than installed version")
}
break
}
break
}
}
return repository, nil
}
func (runtime *tentacleRuntime) prepareManagedPackage(ctx context.Context, repository tentacle.Repository, id string) (*tentacle.PreparedPackage, error) {
prepare := runtime.preparePackage
if prepare == nil {
prepare = tentacle.PreparePackage
}
prepareCtx, cancel := context.WithTimeout(ctx, 45*time.Second)
stop := context.AfterFunc(runtime.lifecycleContext, cancel)
prepared, err := prepare(prepareCtx, tentacle.PackageRoot(runtime.registry.Path), repository, id)
stop()
cancel()
return prepared, err
}
func (runtime *tentacleRuntime) fetchManagedRepository(ctx context.Context, origin string) (tentacle.Repository, error) {
fetch := runtime.fetchRepository
if fetch == nil {
fetch = tentacle.FetchRepository
}
// Fetch on the Preferences worker, leaving lifecycle and native input
// free to disable, revoke, and process failures while the network waits.
fetchCtx, cancel := context.WithTimeout(ctx, 45*time.Second)
stop := context.AfterFunc(runtime.lifecycleContext, cancel)
repository, err := fetch(fetchCtx, origin)
stop()
cancel()
return repository, err
}
+200
View File
@@ -0,0 +1,200 @@
// Copyright (c) 2026 Storm Dragon
// Authors: Storm Dragon, Angus Kola
// SPDX-License-Identifier: GPL-3.0-or-later
package app
import (
"errors"
"slices"
"git.stormux.org/storm/cthulhu/internal/tentacle"
)
func (owner *tentacleLifecycle) ensureManager() error {
if owner.manager != nil {
return nil
}
if owner.runtime.input == nil {
return errors.New("native bindings are unavailable")
}
var err error
owner.manager, err = owner.runtime.makeManager()
return err
}
func (owner *tentacleLifecycle) revoke(id string) {
if session := owner.sessions[id]; session != nil {
session.Disable()
owner.runtime.output.RetireRevoked()
}
}
func (owner *tentacleLifecycle) replyStartup(id string, err error) {
startup, exists := owner.startups[id]
if !exists {
return
}
delete(owner.startups, id)
startup.stop()
if err == nil {
if startup.previous != nil {
previous := *startup.previous
// Starting the replacement already waited for the old session.
owner.removedPackages[previous.PackageDirectory] = tentacleRetiredPackage{entry: previous}
}
err = startup.saveError
}
owner.setSnapshot()
startup.call.result <- err
}
func (owner *tentacleLifecycle) failStartup(id, message string) {
owner.problems[id] = message
startup, exists := owner.startups[id]
if !exists {
return
}
owner.revoke(id)
if owner.current != nil && owner.current.TentacleID() == id {
owner.current, owner.ready = nil, nil
}
delete(owner.pending, id)
next := tentacle.CloneEntries(owner.entries)
for index := range next {
if next[index].ID == id {
if startup.previous != nil {
next[index] = *startup.previous
} else {
next[index].Enabled = false
}
}
}
err := owner.runtime.registry.Save(next)
if err == nil || errors.Is(err, tentacle.ErrRegistrySync) {
if startup.previous != nil {
failed := slices.IndexFunc(owner.entries, func(entry tentacle.Entry) bool { return entry.ID == id })
if failed >= 0 {
entry := owner.entries[failed]
owner.removedPackages[entry.PackageDirectory] = tentacleRetiredPackage{entry: entry, session: owner.sessions[id]}
}
owner.runtime.mutex.Lock()
disabled := owner.runtime.disabled
owner.runtime.mutex.Unlock()
owner.pending[id] = startup.previous.Enabled && !disabled && owner.ctx.Err() == nil
message += " The previous package was restored."
}
owner.entries = next
}
if err != nil {
message += " The recovery state could not be saved durably: " + err.Error()
}
owner.replyStartup(id, errors.New(message))
}
func (owner *tentacleLifecycle) cancelStartups() {
for id := range owner.startups {
owner.failStartup(id, "Tentacle startup was cancelled before activation.")
}
}
func (owner *tentacleLifecycle) startPending() {
for _, entry := range owner.entries {
if !owner.pending[entry.ID] || owner.sessions[entry.ID] != nil {
continue
}
owner.pending[entry.ID] = false
err := owner.ensureManager()
var session *tentacle.Session
if err == nil {
session, err = owner.manager.Start(owner.ctx, entry.Spec())
}
if err != nil {
owner.failStartup(entry.ID, "Tentacle could not start; check resource containment and its program, then enable it again.")
owner.runtime.logger.Warn("start configured tentacle", "tentacle", entry.ID, "error", err)
owner.runtime.report("A tentacle could not start.")
owner.notify()
break
}
owner.sessions[entry.ID] = session
owner.runtime.mutex.Lock()
owner.runtime.sessions = append(owner.runtime.sessions, session)
if owner.runtime.disabled {
session.Disable()
}
owner.runtime.mutex.Unlock()
owner.current, owner.ready = session, session.Ready()
done := make(chan struct{})
// Prune joined watcher handles so retries never accumulate history.
owner.watchers = slices.DeleteFunc(owner.watchers, func(done <-chan struct{}) bool {
select {
case <-done:
return true
default:
return false
}
})
owner.watchers = append(owner.watchers, done)
go func() {
defer close(done)
select {
case <-session.Revoked():
case <-owner.ctx.Done():
}
owner.runtime.output.RetireRevoked()
owner.runtime.signal_extensions()
owner.notify()
<-session.Done()
owner.notify()
}()
break
}
}
func (owner *tentacleLifecycle) activateReady() error {
session := owner.current
owner.current, owner.ready = nil, nil
if state, _ := session.Status(); state != tentacle.Ready {
owner.failStartup(session.TentacleID(), "Tentacle failed during startup; check its program and resource limits, then enable it again.")
return nil
}
if startup, exists := owner.startups[session.TentacleID()]; exists && startup.call.ctx.Err() != nil {
owner.failStartup(session.TentacleID(), "Tentacle startup was cancelled before activation.")
return nil
}
candidate := append(slices.Clone(owner.active), session)
if err := owner.runtime.bind(owner.ctx, candidate); err != nil {
session.Disable()
owner.runtime.logger.Warn("configured tentacle bindings rejected", "tentacle", session.TentacleID(), "error", err)
owner.runtime.report("A tentacle could not acquire its keyboard shortcuts.")
owner.failStartup(session.TentacleID(), "Tentacle could not acquire its keyboard shortcuts; enable it again after resolving the conflict.")
return nil
}
// Native binding setup can wait for held keys. Cancellation during
// that wait must roll back the new grabs before reporting failure.
activationError := owner.ctx.Err()
if startup, exists := owner.startups[session.TentacleID()]; exists && activationError == nil {
activationError = startup.call.ctx.Err()
}
if activationError == nil {
activationError = session.Activate()
}
if activationError != nil {
session.Disable()
if err := owner.runtime.bind(owner.ctx, owner.active); err != nil && owner.ctx.Err() == nil {
select {
case owner.runtime.failures <- err:
default:
}
return err
}
owner.failStartup(session.TentacleID(), "Tentacle stopped before activation; enable it again to retry.")
return nil
}
owner.active = candidate
delete(owner.problems, session.TentacleID())
owner.runtime.publishCommands(owner.active)
owner.replyStartup(session.TentacleID(), nil)
owner.runtime.logger.Info("configured tentacle active", "tentacle", session.TentacleID(), "session", session.Identity())
return nil
}