feat(inmem): finalize attached controllers on release

bus/bus.go

@@ -8,6 +8,26 @@ import (
 	"github.com/aperturerobotics/util/broadcast"
 )
 
+// ControllerCloseError reports a Controller.Close failure.
+//
+// AddController reports this error through its callback. ExecuteController and
+// an AddController attachment failure may include it in their returned error.
+// Callers can distinguish close failures from execution failures with
+// errors.As.
+type ControllerCloseError struct {
+	Err error
+}
+
+// Error returns the controller close failure.
+func (e *ControllerCloseError) Error() string {
+	return "controller close failed: " + e.Err.Error()
+}
+
+// Unwrap returns the underlying close failure.
+func (e *ControllerCloseError) Unwrap() error {
+	return e.Err
+}
+
 // Bus manages running controllers. It has an attached directive controller,
 // which is used to build declarative state requests between controllers.
 type Bus interface {
@@ -20,24 +40,30 @@ type Bus interface {
 	// controllers are added or removed from the bus.
 	GetControllersBroadcast() *broadcast.Broadcast
 
-	// AddController adds a controller to the bus and calls Execute().
-	// The controller will exit if ctx is canceled.
-	// Returns a release function for the controller reference.
-	// The controller will receive directive callbacks until removed.
-	// Any fatal error in the controller is written to cb.
-	// If the controller is released, cb will be called with nil.
-	// cb can be nil
+	// AddController attaches a controller and calls Execute asynchronously.
+	// The controller receives directive callbacks until it is released,
+	// Execute returns an error, or Execute panics.
+	//
+	// The returned release function cancels the Execute context, detaches
+	// directive handling, waits for Execute to return, calls Close once, calls
+	// cb, and then returns. Repeated release calls are idempotent.
+	//
+	// cb receives the execution error, a ControllerCloseError, both errors
+	// joined together, or nil when release completes without either error.
+	// cb can be nil.
 	AddController(ctx context.Context, ctrl controller.Controller, cb func(exitErr error)) (func(), error)
 
-	// ExecuteController adds a controller to the bus and calls Execute().
-	// The controller will exit if ctx is canceled.
-	// Any fatal error in the controller is returned.
-	// The controller will receive directive callbacks.
-	// If this function returns nil, call RemoveController to remove the controller.
+	// ExecuteController attaches a controller and calls Execute synchronously.
+	// A nil Execute return leaves the controller attached until
+	// RemoveController is called. An Execute error or panic detaches the
+	// controller, calls Close once, and is returned, joined with any
+	// ControllerCloseError.
+	// RemoveController may be called concurrently to cancel and finalize the
+	// synchronous execution.
 	ExecuteController(context.Context, controller.Controller) error
 
-	// RemoveController removes the controller from the bus.
-	// The controller will no longer receive callbacks.
-	// Note: this might not cancel the Execute() context automatically.
+	// RemoveController synchronously releases one attached instance of the
+	// controller. It cancels that instance's Execute context, detaches
+	// directive handling, waits for Execute to return, and calls Close once.
 	RemoveController(controller.Controller)
 }

bus/inmem/controller.go

@@ -1,11 +1,88 @@
 package inmem
 
-import "github.com/aperturerobotics/controllerbus/controller"
+import (
+	"context"
+	"errors"
+	"sync"
+	"time"
 
-// attachedCtrl contains an attached controller
+	"github.com/aperturerobotics/controllerbus/bus"
+	"github.com/aperturerobotics/controllerbus/controller"
+)
+
+const releaseWarningInterval = 30 * time.Second
+
+// attachedCtrl contains the lifecycle of one attached controller instance.
 type attachedCtrl struct {
-	// ctrl is the controller
 	ctrl controller.Controller
-	// rel releases the controller
-	rel func()
+	rel  func()
+
+	cancel      context.CancelFunc
+	executeDone chan struct{}
+	executeErr  error
+	callback    func(error)
+
+	finalizeOnce sync.Once
+	finalErr     error
+}
+
+// newAttachedCtrl constructs an attached controller lifecycle.
+func newAttachedCtrl(
+	ctrl controller.Controller,
+	rel func(),
+	cancel context.CancelFunc,
+	callback func(error),
+) *attachedCtrl {
+	return &attachedCtrl{
+		ctrl:        ctrl,
+		rel:         rel,
+		cancel:      cancel,
+		executeDone: make(chan struct{}),
+		callback:    callback,
+	}
+}
+
+// finishExecution records the result before publishing execution completion.
+func (c *attachedCtrl) finishExecution(err error) {
+	c.executeErr = err
+	close(c.executeDone)
+}
+
+// finalize cancels, detaches, waits, closes, and reports exactly once.
+func (c *attachedCtrl) finalize(b *Bus) error {
+	c.finalizeOnce.Do(func() {
+		c.cancel()
+		b.detachController(c)
+
+		ticker := time.NewTicker(releaseWarningInterval)
+		defer ticker.Stop()
+	waitForExecute:
+		for {
+			select {
+			case <-c.executeDone:
+				break waitForExecute
+			case <-ticker.C:
+				b.le.WithField("controller", c.ctrl).Warn("waiting for controller Execute to return")
+			}
+		}
+
+		c.finalErr = joinControllerErrors(c.executeErr, c.ctrl.Close())
+		if c.callback != nil {
+			c.callback(c.finalErr)
+		}
+	})
+	return c.finalErr
+}
+
+// joinControllerErrors preserves the execution error and identifies a close
+// failure for errors.As.
+func joinControllerErrors(executeErr, closeErr error) error {
+	if closeErr == nil {
+		return executeErr
+	}
+	wrappedCloseErr := &bus.ControllerCloseError{Err: closeErr}
+	if executeErr == nil {
+		return wrappedCloseErr
+	}
+	return errors.Join(executeErr, wrappedCloseErr)
 }

bus/inmem/detach_test.go

@@ -0,0 +1,61 @@
+package inmem
+
+import "testing"
+
+func TestDetachControllerIsIdempotent(t *testing.T) {
+	releaseCalls := 0
+	target := &attachedCtrl{
+		rel: func() { releaseCalls++ },
+	}
+	retained := &attachedCtrl{}
+	b := NewBus(nil)
+	b.controllers = []*attachedCtrl{target, retained}
+
+	locked := b.bcast.Lock()
+	firstBroadcast := locked.WaitCh()
+	locked.Unlock()
+
+	b.detachController(target)
+
+	if got := releaseCalls; got != 1 {
+		t.Fatalf("release calls after first detach = %d, want 1", got)
+	}
+	if got := len(b.controllers); got != 1 {
+		t.Fatalf("controller count after first detach = %d, want 1", got)
+	}
+	if b.controllers[0] != retained {
+		t.Fatal("retained controller changed after first detach")
+	}
+
+	broadcastCalls := 0
+	select {
+	case <-firstBroadcast:
+		broadcastCalls++
+	default:
+		t.Fatal("first detach did not broadcast")
+	}
+
+	locked = b.bcast.Lock()
+	secondBroadcast := locked.WaitCh()
+	locked.Unlock()
+
+	b.detachController(target)
+
+	if got := releaseCalls; got != 1 {
+		t.Fatalf("release calls after second detach = %d, want 1", got)
+	}
+	if got := len(b.controllers); got != 1 {
+		t.Fatalf("controller count after second detach = %d, want 1", got)
+	}
+	if b.controllers[0] != retained {
+		t.Fatal("retained controller changed after second detach")
+	}
+	select {
+	case <-secondBroadcast:
+		broadcastCalls++
+	default:
+	}
+	if broadcastCalls != 1 {
+		t.Fatalf("broadcast calls after two detaches = %d, want 1", broadcastCalls)
+	}
+}