amadan

carlos / platform Public

activator: a run's lost start is settled, a stop does not wait for Prepare, runs are for instances (review round 1)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude e5abb4dc19fb32138edfd88b601d18b3548befed parent f5dacb4
4 files changed, +267 −31
  • internal/activator/activator.go +6 −1
  • internal/activator/cmd.go +97 −27
  • internal/activator/cmd_test.go +153 −0
  • internal/activator/cmdunit.go +11 −3

CI passed — run details

diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index c1bcc3731..3fd5980d1 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -1394,6 +1394,9 @@ const actStopTimeout = 20 * time.Second
func (a *Service) beginStopLocked(e *actEntry, why string) {
e.state = actStopping
e.stopWhy = why
+ if e.cmd != nil {
+ e.cmd.cancelPrepare() // a stop interrupts a run's download
+ }
e.stopped = make(chan struct{})
go a.stop(e)
}
@@ -2383,7 +2386,9 @@ func (a *Service) finishStop(e *actEntry, outcome *stopObservation) {
// A run that will not settle leaves the stop unproven.
a.mu.Lock()
holds := append([]*cmdHold(nil), e.cmdLeft...)
- if e.cmd != nil {
+ // A run not yet starting is not waited for: RunCmd's last awake check
+ // sees this stop and abandons it before any unit starts.
+ if e.cmd != nil && e.cmd.starting {
holds = append(holds, e.cmd)
}
why := e.stopWhy
diff --git a/internal/activator/cmd.go b/internal/activator/cmd.go
index 87f7ba670..382a2e9ee 100644
--- a/internal/activator/cmd.go
+++ b/internal/activator/cmd.go
@@ -28,9 +28,14 @@ const (
// root protocol's reservations (#250), so they are refused (run spec §3.1).
var ErrCmdClaim = errors.New("activator: claim control is armed; runs wait for the reservation protocol (#250)")
-// ErrCmdBusy: a run is already attached to the instance.
+// ErrCmdBusy: a run is already attached to the instance, or an earlier
+// run's unit was never proven settled.
var ErrCmdBusy = errors.New("activator: a run is already attached to this instance")
+// ErrCmdNotInstance: runs are for instances; the route is a worker or
+// another kind.
+var ErrCmdNotInstance = errors.New("activator: runs are only for instances, and this route is not one")
+
// CmdWakeError: the wake failed or did not finish within cmdWakeBound.
// Reason is member-safe (memberSafeWakeReason).
type CmdWakeError struct {
@@ -53,10 +58,17 @@ func (e *CmdStoppedError) Error() string {
var cmdWakeBound = 60 * time.Second
// cmdStopBound bounds finishStop's wait for a run: the start in flight
-// (DeadlineControl), the stop and the cgroup (actStopTimeout) and both
-// streams (cmdDrainBound), with room.
+// (cmdStartBound), the stop (DeadlineControl), the cgroup (actStopTimeout)
+// and both streams (cmdAcceptGrace + cmdDrainBound each), with room.
+// Shutdown runs inside systemd's 90 s TimeoutStopSec for the agent: a
+// settle (about 62 s) fits; a start in flight on top may not, and an agent
+// killed mid-stop leaves the lease to its TTL.
const cmdStopBound = 2 * time.Minute
+// cmdStartBound bounds RunCmd's StartCmd, well below cmdStopBound, so a
+// stop waiting on the start still has the settle's budget after it.
+const cmdStartBound = 45 * time.Second
+
// CmdHandle is a started run, as the box's run pass sees it.
type CmdHandle interface {
Invocation() string
@@ -74,19 +86,26 @@ type CmdHandle interface {
// cmdHold is the entry's cmd field.
type cmdHold struct {
- runID string
- started chan struct{} // closed once StartCmd returned, either way
- unit CmdUnit // written before started is closed; nil on a failed start
- mu sync.Mutex
- asked bool // a stop has been asked for; under mu
- reason string
- stopOnce sync.Once
- settled chan struct{}
- stopErr error
+ runID string
+ // starting is set under a.mu by RunCmd's last awake check: from then
+ // on a stop waits for the start. Before it, a stop does not wait: the
+ // check will see the entry stopping and abandon the run.
+ starting bool
+ // cancelPrepare cancels spec.Prepare's context; beginStopLocked calls
+ // it, under a.mu, so a stop interrupts a download.
+ cancelPrepare context.CancelFunc
+ started chan struct{} // closed once StartCmd returned, either way
+ unit CmdUnit // written before started is closed; nil when no unit can be running
+ mu sync.Mutex
+ asked bool // a stop has been asked for; under mu
+ reason string
+ stopOnce sync.Once
+ settled chan struct{}
+ stopErr error
}
-func newCmdHold(runID string) *cmdHold {
- return &cmdHold{runID: runID, started: make(chan struct{}), settled: make(chan struct{})}
+func newCmdHold(runID string, cancelPrepare context.CancelFunc) *cmdHold {
+ return &cmdHold{runID: runID, cancelPrepare: cancelPrepare, started: make(chan struct{}), settled: make(chan struct{})}
}
// stop waits for the start to return, records reason if it is the first
@@ -122,7 +141,9 @@ func (h *cmdHold) stop(ctx context.Context, reason string) error {
}
// provenSettled reports whether the hold leaves nothing running: the start
-// returned no unit, or the unit settled without error.
+// returned no unit, or the unit settled without error. A failed start
+// that reached the control service returns a unit, since the unit may be
+// running.
func (h *cmdHold) provenSettled() bool {
select {
case <-h.started:
@@ -177,18 +198,38 @@ func (r *cmdRun) Release() {
})
}
+// cmdBusyLocked reports whether e cannot take a run: one is attached, or
+// an earlier one's unit was never proven settled. It drops the released
+// holds that have since settled. The caller holds a.mu.
+func cmdBusyLocked(e *actEntry) bool {
+ left := e.cmdLeft[:0]
+ for _, h := range e.cmdLeft {
+ if !h.provenSettled() {
+ left = append(left, h)
+ }
+ }
+ clear(e.cmdLeft[len(left):])
+ e.cmdLeft = left
+ return e.cmd != nil || len(left) > 0
+}
+
// RunCmd runs one command against rt's instance (run spec §5.2 and §5.3):
-// refused with claim control armed or on a layout other than nested; the
-// instance woken through the ordinary wake path, waiting up to 60 s; the
-// run attached to the awake entry; spec.Prepare run under the hold; and
+// refused with claim control armed, for anything but an instance, or on a
+// layout other than nested; the instance woken through the ordinary wake
+// path, waiting up to 60 s; the run attached to the awake entry;
+// spec.Prepare run under the hold, its context cancelled by a stop; and
// the unit started only while the entry is still awake. A stop that begins
-// after the attach is either seen here (CmdStoppedError, no unit) or
-// waits in finishStop for the start to return and then stops the unit
-// before replication closes.
+// before the last awake check does not wait for the run, which then ends
+// here (CmdStoppedError, no unit). One that begins after it waits in
+// finishStop for the start to return and then stops the unit before
+// replication closes.
func (a *Service) RunCmd(ctx context.Context, rt registry.Route, spec CmdSpec) (CmdHandle, error) {
if a.cfg.HostWrites.Enabled {
return nil, ErrCmdClaim
}
+ if rt.Kind != "instance" {
+ return nil, ErrCmdNotInstance
+ }
cb, ok := a.backend.(CmdBackend)
if !ok {
return nil, ErrCmdNotNested
@@ -215,7 +256,9 @@ func (a *Service) RunCmd(ctx context.Context, rt registry.Route, spec CmdSpec) (
}
return nil, &CmdWakeError{Reason: memberSafeWakeReason(err), Err: err}
}
- h := newCmdHold(spec.RunID)
+ pctx, pcancel := context.WithCancel(ctx)
+ defer pcancel()
+ h := newCmdHold(spec.RunID, pcancel)
a.mu.Lock()
e := a.entries[rt.Host]
switch {
@@ -227,7 +270,11 @@ func (a *Service) RunCmd(ctx context.Context, rt registry.Route, spec CmdSpec) (
a.mu.Unlock()
release()
return nil, &CmdStoppedError{Reason: why}
- case e.cmd != nil:
+ case e.worker:
+ a.mu.Unlock()
+ release()
+ return nil, ErrCmdNotInstance
+ case cmdBusyLocked(e):
a.mu.Unlock()
release()
return nil, ErrCmdBusy
@@ -241,10 +288,9 @@ func (a *Service) RunCmd(ctx context.Context, rt registry.Route, spec CmdSpec) (
run.Release()
return nil, err
}
+ var prepErr error
if spec.Prepare != nil {
- if err := spec.Prepare(ctx); err != nil {
- return abandon(err)
- }
+ prepErr = spec.Prepare(pctx)
}
a.mu.Lock()
if e.state != actAwake {
@@ -252,13 +298,37 @@ func (a *Service) RunCmd(ctx context.Context, rt registry.Route, spec CmdSpec) (
a.mu.Unlock()
return abandon(&CmdStoppedError{Reason: why})
}
+ if prepErr != nil {
+ a.mu.Unlock()
+ return abandon(prepErr)
+ }
+ h.starting = true
a.mu.Unlock()
- u, err := cb.StartCmd(ctx, rt, spec)
+ sctx, scancel := context.WithTimeout(ctx, cmdStartBound)
+ u, err := cb.StartCmd(sctx, rt, spec)
+ scancel()
h.unit = u
close(h.started)
if err != nil {
+ if u != nil {
+ // The start reached the control service, so the unit may be
+ // running: settle it now. Release keeps it for the next stop
+ // if it does not settle.
+ sctx, scancel := context.WithTimeout(context.WithoutCancel(ctx), cmdStopBound)
+ _ = h.stop(sctx, "")
+ scancel()
+ }
run.Release()
return nil, err
}
+ a.mu.Lock()
+ stopping, why := e.state != actAwake, e.stopWhy
+ a.mu.Unlock()
+ if stopping {
+ // A stop began during the start. finishStop is waiting to stop the
+ // unit, unless its wait already timed out: stop it here too (once
+ // either way), so a late start never outlives the entry.
+ go h.stop(context.Background(), why) //nolint:errcheck
+ }
return run, nil
}
diff --git a/internal/activator/cmd_test.go b/internal/activator/cmd_test.go
index a8eca9ef2..a7f61c5b7 100644
--- a/internal/activator/cmd_test.go
+++ b/internal/activator/cmd_test.go
@@ -4,12 +4,15 @@ import (
"context"
"errors"
"path/filepath"
+ "strings"
"sync"
+ "sync/atomic"
"testing"
"time"
"github.com/carlosframework/platform/internal/hostwrite"
"github.com/carlosframework/platform/internal/instancectl"
+ "github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
)
@@ -76,6 +79,11 @@ type cmdFakeBackend struct {
cmdStarts int
gate chan struct{} // when set, StartCmd blocks until it closes
entered chan struct{} // when set, closed as StartCmd is entered
+ // errWithUnit makes a failing start return the unit too: the start
+ // reached the control service and the unit may be running.
+ errWithUnit bool
+ // startCmd, when set, replaces the fake start (a real ExecBackend's).
+ startCmd func(context.Context, registry.Route, CmdSpec) (CmdUnit, error)
}
func (b *cmdFakeBackend) StartCmd(ctx context.Context, rt registry.Route, spec CmdSpec) (CmdUnit, error) {
@@ -88,7 +96,13 @@ func (b *cmdFakeBackend) StartCmd(ctx context.Context, rt registry.Route, spec C
b.cmdMu.Lock()
defer b.cmdMu.Unlock()
b.cmdStarts++
+ if b.startCmd != nil {
+ return b.startCmd(ctx, rt, spec)
+ }
if b.startErr != nil {
+ if b.errWithUnit {
+ return b.unit, b.startErr
+ }
return nil, b.startErr
}
return b.unit, nil
@@ -359,3 +373,142 @@ func TestAReleasedRunThatNeverSettledStillHoldsTheStop(t *testing.T) {
})
}
}
+
+func TestRunCmdRefusesAnythingButAnInstance(t *testing.T) {
+ a, cb, _ := testCmdActivator(t, Config{})
+ rt := testRoute("c.example.com")
+ rt.Kind = "sidecar"
+ if _, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"}); !errors.Is(err, ErrCmdNotInstance) {
+ t.Fatalf("err = %v, want ErrCmdNotInstance", err)
+ }
+ if acquires, _, _ := cb.counts(); acquires != 0 {
+ t.Fatal("a refused route was woken")
+ }
+}
+
+// A stop during a slow Prepare (the file downloads) does not wait for it:
+// no unit has started, so the stop is prompt and complete, Prepare's
+// context is cancelled, and the run ends interrupted with no unit started.
+func TestAStopDuringASlowPrepareDoesNotWaitForIt(t *testing.T) {
+ a, cb, _ := testCmdActivator(t, Config{})
+ rt := testRoute("c.example.com")
+ inPrepare, unblock := make(chan struct{}), make(chan struct{})
+ var cancelled atomic.Bool
+ done := make(chan error, 1)
+ go func() {
+ _, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a", Prepare: func(ctx context.Context) error {
+ close(inPrepare)
+ <-unblock // a download that does not watch its context
+ cancelled.Store(ctx.Err() != nil)
+ return nil
+ }})
+ done <- err
+ }()
+ <-inPrepare
+ a.Bounce(rt.Host)
+ waitFor(t, "the instance stopped while Prepare was still running", func() bool { return a.entryCount() == 0 })
+ if a.StopUncertain(rt.Host) {
+ t.Fatal("a stop that no unit outlived was recorded uncertain")
+ }
+ close(unblock)
+ var se *CmdStoppedError
+ if err := <-done; !errors.As(err, &se) || se.Reason != StopRestart {
+ t.Fatalf("err = %v, want CmdStoppedError{%q}", err, StopRestart)
+ }
+ if !cancelled.Load() {
+ t.Error("the stop did not cancel Prepare's context")
+ }
+ if cb.cmdStarts != 0 {
+ t.Fatal("a unit started after the stop")
+ }
+}
+
+// A start that failed after it reached the control service may have left
+// the unit running: RunCmd settles it, and a unit that will not settle
+// keeps the instance from another run and makes its next stop uncertain.
+func TestAFailedStartThatMayHaveRunIsSettled(t *testing.T) {
+ for _, c := range []struct {
+ name string
+ settleErr error
+ }{{"settles", nil}, {"will not settle", errors.New("processes remained")}} {
+ t.Run(c.name, func(t *testing.T) {
+ a, cb, _ := testCmdActivator(t, Config{})
+ cb.startErr, cb.errWithUnit = errors.New("502 mid-roll"), true
+ cb.unit.settleErr = c.settleErr
+ rt := testRoute("c.example.com")
+ if _, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"}); !errors.Is(err, cb.startErr) {
+ t.Fatalf("err = %v, want the start's error", err)
+ }
+ if stops, _ := cb.unit.stopped(); stops != 1 {
+ t.Fatalf("the unit was stopped %d times, want 1", stops)
+ }
+ _, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_b"})
+ if busy := errors.Is(err, ErrCmdBusy); busy != (c.settleErr != nil) {
+ t.Fatalf("second run: err = %v, busy = %v", err, busy)
+ }
+ a.Bounce(rt.Host)
+ waitFor(t, "the instance stopped", func() bool { return a.entryCount() == 0 })
+ if _, repWasClosed := cb.unit.stopped(); repWasClosed {
+ t.Fatal("replication closed before the unit settled")
+ }
+ if got := a.StopUncertain(rt.Host); got != (c.settleErr != nil) {
+ t.Fatalf("StopUncertain = %v", got)
+ }
+ })
+ }
+}
+
+// The real ExecBackend: POST /cmd starts the unit and then fails (a lost
+// response). The cmd role is stopped and its cgroup waited for before
+// replication closes; a cgroup that cannot be read empty leaves the stop
+// uncertain.
+func TestALostCmdResponseStopsTheCmdRoleBeforeReplicationCloses(t *testing.T) {
+ for _, c := range []struct {
+ name string
+ cgroupErr error
+ }{{"cgroup empties", nil}, {"cgroup unreadable", errors.New("cgroup.events: permission denied")}} {
+ t.Run(c.name, func(t *testing.T) {
+ a, cb, _ := testCmdActivator(t, Config{})
+ ctl := &fakeControl{cmdErr: errors.New("502 mid-roll")}
+ unit := &fakeUnit{}
+ unit.set("active", "inv1", true, true) // the POST reached systemd
+ b, rt := nestedBackend(t, ctl, unit)
+ b.layoutFn = func(registry.Route) (Layout, string) { return LayoutNested, "" }
+ var repClosedAtEmpty atomic.Bool
+ b.CgroupEvents = func(slice, u string) (bool, bool, error) {
+ if c.cgroupErr != nil {
+ return true, true, c.cgroupErr
+ }
+ exists, populated, err := unit.events(slice, u)
+ if !populated {
+ repClosedAtEmpty.Store(cb.lastRep().isClosed())
+ }
+ return exists, populated, err
+ }
+ ctl.stopHook = func(_ string, role instancectl.Role) {
+ if role == instancectl.RoleCmd {
+ go func() {
+ time.Sleep(50 * time.Millisecond)
+ unit.set("inactive", "inv1", false, false)
+ }()
+ }
+ }
+ cb.startCmd = b.StartCmd
+ dir := filepath.Join(provision.InstanceDir(b.DataDir, rt.Host), CmdRunDirName, "run_a")
+ if _, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a", RunDir: dir}); err == nil || !strings.Contains(err.Error(), "502 mid-roll") {
+ t.Fatalf("err = %v, want the lost response", err)
+ }
+ a.Bounce(rt.Host)
+ waitFor(t, "the instance stopped", func() bool { return a.entryCount() == 0 })
+ if !strings.Contains(ctl.joined(), "stop "+rt.Host+" cmd") {
+ t.Fatalf("the cmd role was never stopped:\n%s", ctl.joined())
+ }
+ if c.cgroupErr == nil && repClosedAtEmpty.Load() {
+ t.Fatal("replication closed before the cmd cgroup emptied")
+ }
+ if got := a.StopUncertain(rt.Host); got != (c.cgroupErr != nil) {
+ t.Fatalf("StopUncertain = %v", got)
+ }
+ })
+ }
+}
diff --git a/internal/activator/cmdunit.go b/internal/activator/cmdunit.go
index bf84cf9e8..36d6d5442 100644
--- a/internal/activator/cmdunit.go
+++ b/internal/activator/cmdunit.go
@@ -59,7 +59,9 @@ type CmdUnit interface {
StopAndSettle(ctx context.Context) error
}
-// CmdBackend is the backend capability Service.RunCmd needs.
+// CmdBackend is the backend capability Service.RunCmd needs. StartCmd may
+// return a unit together with an error: the start failed after the unit
+// may have begun, and the unit must still be stopped and settled.
type CmdBackend interface {
StartCmd(ctx context.Context, rt registry.Route, spec CmdSpec) (CmdUnit, error)
}
@@ -110,7 +112,9 @@ func (b *ExecBackend) cmdLayout(rt registry.Route) (Layout, string) {
// StartCmd starts one run's cmd unit (run spec §5.4): the layout check, the
// nested gate, the live link, the instance's directories and environment,
// the two stream listeners, launch.cmd.json, then POST /cmd. It returns
-// once the unit has an invocation. The environment is never logged.
+// once the unit has an invocation. The environment is never logged. A
+// failure after the POST was sent returns a unit with the error: the unit
+// may be running, and the caller must stop and settle it.
func (b *ExecBackend) StartCmd(ctx context.Context, rt registry.Route, spec CmdSpec) (CmdUnit, error) {
if l, why := b.cmdLayout(rt); l != LayoutNested {
return nil, fmt.Errorf("%w: %s is %s (%s)", ErrCmdNotNested, rt.Host, l, why)
@@ -149,7 +153,11 @@ func (b *ExecBackend) StartCmd(ctx context.Context, rt registry.Route, spec CmdS
}
inv, err := b.Control.Cmd(ctx, rt.Host, b.tier(rt).MemoryMax)
if err != nil {
- return fail(fmt.Errorf("%s: start cmd unit: %w", rt.Host, err))
+ // The request may have reached systemd before it failed (a timeout,
+ // a lost response, a 502 mid-roll), so the unit may be running.
+ // Hand back a unit whose StopAndSettle stops the host's cmd role
+ // and waits for its cgroup; it needs no invocation.
+ return &cmdUnit{b: b, rt: rt, out: out, errs: errs}, fmt.Errorf("%s: start cmd unit: %w", rt.Host, err)
}
log.Printf("activator: %s: started %s for run %s", rt.Host, instancectl.UnitName(instancectl.RoleCmd, rt.Host), spec.RunID)
return &cmdUnit{b: b, rt: rt, inv: inv, out: out, errs: errs}, nil