carlos / platform Public

activator: recheck the exec after the wake record; nested seed obligation first (#272)

Astra's review of zdd-owed-items f4097c6c.

P1: the pending record is a bucket write, and the lease, a fence or the
attempt can change while it runs. The exec is now decided again after it:
the pending record asks the wake's own admission (its context and the
maintenance fence) after its write, and a seam with an exec gate
(runExecGate, launchGateWith) asks the gate, and the live check, again once
the write has run. A refusal there leaves the record naming a build that
did not run. It is not undone: an undo is a second bucket write after the
lease may be lost, which the build state's single-writer rule forbids, and
the window is the one main had before this branch (record at step 2b, then
the fence check). The reasoning is on pendingRecord.

P2: a nested seed wrote its record before its obligation, so an obligation
that failed left a record for a seed that never ran. The obligation now
comes first; a record that fails, or a recheck that refuses, discharges
the obligation just written and keeps the database.

Tests (each fails with its fix taken out):
TestAnExecGateRefusingDuringTheRecordStartsNothing,
TestAFenceRaisedDuringTheRecordStartsNothing,
TestANestedSeedWritesItsObligationBeforeItsRecord.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

claude 1 week ago d134fc12595214616a2a968d51b1d1aa9516bd57 parent f4097c6

6 files changed, +280 −37

internal/activator/backend_exec.go +13 −6
internal/activator/beforeexec.go +62 −16
internal/activator/beforeexec_test.go +4 −4
internal/activator/genbackend.go +32 −10
internal/activator/wakeidentity_test.go +1 −1
internal/activator/wakerecord_recheck_test.go +168 −0

diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go

index 9ab17047e..0843a90bd 100644

--- a/internal/activator/backend_exec.go

+++ b/internal/activator/backend_exec.go

@@ -1487,17 +1487,24 @@ func (b *ExecBackend) seedUnit(ctx context.Context, rt registry.Route, exe *inst

// unchanged for the unit path.

_, statErr := os.Stat(dbPath)

seedOwnsDB := seedCreatedTheDB(statErr)

- // The wake's pending record, once every refusal on this side has passed

- // (amadan #272), and before the obligation, so a record that fails

- // leaves nothing to discharge.

- if err := runPendingRecord(ctx); err != nil {

- return err

- }

// The obligation is written ahead, as seedChild's is, before root is

// asked to run anything; it names the seed unit as the proof to ask.

if err := recordSeedObligation(host, dbPath, seedObligation{ownsDB: seedOwnsDB, layout: LayoutNested}); err != nil {

return err

}

+ // Then the wake's pending record (amadan #272), the last step before

+ // root is asked, as the flat seed's comes at its exec after its

+ // obligation (Astra review of zdd-owed-items, P2): an obligation that

+ // cannot be written records nothing. A record that fails, or a recheck

+ // that refuses after it, runs no seed, so the obligation just written

+ // is discharged and the database, which no seed touched, is kept.

+ if _, err := runPendingRecord(ctx); err != nil {

+ cause := fmt.Errorf("seed for %s: %w", host, err)

+ if derr := dischargeSeedObligation(dbPath, false); derr != nil {

+ return fmt.Errorf("%w; and %w", cause, derr)

+ }

+ return cause

+ }

log.Printf("activator: seeding %s: %s", host, strings.Join(argv, " "))

// The client already bounds a seed call at instancectl.DeadlineSeed.

diff --git a/internal/activator/beforeexec.go b/internal/activator/beforeexec.go

index 965376e57..4b7219665 100644

--- a/internal/activator/beforeexec.go

+++ b/internal/activator/beforeexec.go

@@ -207,14 +207,44 @@ func (a *Service) settleWakeIdentity(ctx context.Context, ib identityBackend, e

rctx = withRecoverySelection(ctx, plan, rt, id.Version)

}

build := id.Version

- return id, pend, withPendingRecord(ctx, build, func() error { return a.recordBuild(rctx, rt.Host, build) }), nil

+ write := func() error { return a.recordBuild(rctx, rt.Host, build) }

+ // The wake's own admission, asked again once the write is done (Astra

+ // review of zdd-owed-items, P1): the bucket write takes time, and a

+ // fence raised or the wake's context ended meanwhile must start

+ // nothing, as doWake's own check before the spawn refuses it.

+ recheck := func() error {

+ if ctx.Err() != nil || !a.wakeAllowed(rt.Host) {

+ return ErrFenced

+ }

+ return nil

+ }

+ return id, pend, withPendingRecord(ctx, build, write, recheck), nil

}

// pendingRecord is one wake's record of build, written at most once, at

// the first exec that asks for it (runPendingRecord).

+//

+// The write is a bucket PUT, and the lease can be lost, a fence raised or

+// an attempt superseded while it runs. So every exec that consults the

+// record asks recheck (the wake's own admission: its context and the

+// maintenance fence) after it, and an exec seam that has an exec gate asks

+// that gate again once the write has run (runExecGate, launchGateWith).

+// Either refusal starts nothing.

+//

+// A refusal there leaves the record naming a build that did not run. It is

+// not undone. An undo would be a second bucket write, made after the lease

+// may already be lost, so it could land under another holder's lease; the

+// object's CAS would stop it overwriting that holder's record, but writing

+// outside the lease is exactly what the build state's single-writer rule

+// (ruling R-BS-UPDATE) forbids. The window is the one main had before

+// amadan #272, where step 2b wrote the record and doWake then checked the

+// fence: the record names a build that passed every refusal of its own and

+// was about to run, the next wake records again, and a request it completed

+// was the request for that build.

type pendingRecord struct {

- build string

- write func() error

+ build string

+ write func() error

+ recheck func() error

mu sync.Mutex

// done is set once the write ran; err is its answer, which every later

@@ -229,9 +259,9 @@ type pendingRecord struct {

type pendingWakeKey struct{}

// withPendingRecord returns ctx carrying build's pending record, written by

-// write.

-func withPendingRecord(ctx context.Context, build string, write func() error) context.Context {

- return context.WithValue(ctx, pendingWakeKey{}, &pendingRecord{build: build, write: write})

+// write and followed by recheck (nil asks nothing) at every exec.

+func withPendingRecord(ctx context.Context, build string, write, recheck func() error) context.Context {

+ return context.WithValue(ctx, pendingWakeKey{}, &pendingRecord{build: build, write: write, recheck: recheck})

}

func pendingRecordFrom(ctx context.Context) *pendingRecord {

@@ -255,29 +285,42 @@ func (p *pendingRecord) defers(build string) bool {

return !p.superseded

}

-// run writes the record once. A superseded record writes nothing.

-func (p *pendingRecord) run() error {

+// run writes the record once, then asks recheck. wrote reports that this

+// call made the write. A superseded record writes and asks nothing.

+func (p *pendingRecord) run() (wrote bool, err error) {

p.mu.Lock()

- defer p.mu.Unlock()

if p.superseded {

- return nil

+ p.mu.Unlock()

+ return false, nil

}

if !p.done {

- p.err, p.done = p.write(), true

+ p.err, p.done, wrote = p.write(), true, true

+ }

+ err = p.err

+ p.mu.Unlock()

+ if err != nil {

+ return wrote, err

+ }

+ if p.recheck != nil {

+ if err := p.recheck(); err != nil {

+ return wrote, err

+ }

}

- return p.err

+ return wrote, nil

}

// runPendingRecord writes ctx's pending wake record, if it carries one that

// is still due. Every exec seam a cold wake reaches asks it immediately

// before its exec: the flat seed's and spawn's cgroup before-start hook,

// a worker's, and, nested, right before root is asked to run the seed or

-// the run unit. An error refuses that exec, with nothing run.

-func runPendingRecord(ctx context.Context) error {

+// the run unit. An error refuses that exec, with nothing run. wrote

+// reports that this call made the write, so a seam with an exec gate asks

+// it again.

+func runPendingRecord(ctx context.Context) (wrote bool, err error) {

if p := pendingRecordFrom(ctx); p != nil {

return p.run()

}

- return nil

+ return false, nil

}

// pendingRecordHook is runPendingRecord as a cgroup before-start hook for

@@ -287,5 +330,8 @@ func pendingRecordHook(ctx context.Context) func() error {

if pendingRecordFrom(ctx) == nil {

return nil

}

- return func() error { return runPendingRecord(ctx) }

+ return func() error {

+ _, err := runPendingRecord(ctx)

+ return err

+ }

}

diff --git a/internal/activator/beforeexec_test.go b/internal/activator/beforeexec_test.go

index 7ed397f31..6c90b47a6 100644

--- a/internal/activator/beforeexec_test.go

+++ b/internal/activator/beforeexec_test.go

@@ -121,7 +121,7 @@ func TestARecordedBuildIsNotWrittenTwice(t *testing.T) {

t.Fatalf("calls = %v before any exec, want none", calls)

}

for range 2 {

- if err := runPendingRecord(ctx); err != nil {

+ if _, err := runPendingRecord(ctx); err != nil {

t.Fatal(err)

}

@@ -138,7 +138,7 @@ func TestARecordedBuildIsNotWrittenTwice(t *testing.T) {

if err := a.recordBuild(ctx, rt.Host, "v2"); err != nil {

t.Fatal(err)

}

- if err := runPendingRecord(ctx); err != nil {

+ if _, err := runPendingRecord(ctx); err != nil {

t.Fatal(err)

}

if calls, _, _ := s.seen(); !slices.Equal(calls, []string{"one.example.com@v2"}) {

@@ -150,7 +150,7 @@ func TestARecordedBuildIsNotWrittenTwice(t *testing.T) {

a.BeforeExec = f.hook

ctx = wakeRecord(a, rt.Host, "v1")

for range 2 {

- if err := runPendingRecord(ctx); !errors.Is(err, errBuildStateUnwritten) {

+ if _, err := runPendingRecord(ctx); !errors.Is(err, errBuildStateUnwritten) {

t.Fatalf("err = %v, want errBuildStateUnwritten", err)

}

@@ -162,7 +162,7 @@ func TestARecordedBuildIsNotWrittenTwice(t *testing.T) {

// wakeRecord is the context settleWakeIdentity returns for a wake of host

// that settled build: its record pending.

func wakeRecord(a *Service, host, build string) context.Context {

- return withPendingRecord(context.Background(), build, func() error { return a.recordBuild(context.Background(), host, build) })

+ return withPendingRecord(context.Background(), build, func() error { return a.recordBuild(context.Background(), host, build) }, nil)

}

// Ruling R-WAKEWORDS: a member never sees the internal text.

diff --git a/internal/activator/genbackend.go b/internal/activator/genbackend.go

index 9ac8eaac7..4daef066e 100644

--- a/internal/activator/genbackend.go

+++ b/internal/activator/genbackend.go

@@ -85,17 +85,25 @@ func withExecGate(ctx context.Context, gate func() error) context.Context {

}

// runExecGate asks ctx's exec gate, if it carries one, then writes ctx's

-// pending wake record, if it carries one (runPendingRecord, amadan #272):

-// the gate is the last refusal before the exec, and the record is written

-// only once nothing is left to refuse it. A spawn with a gate also honours

-// its context there: one that has ended starts nothing. With neither (every

-// live-link spawn, the flag off) it is a no-op, so those spawns are exactly

-// as they were.

+// pending wake record, if it carries one (runPendingRecord, amadan #272),

+// and, once that write has run, asks the gate again (Astra review of

+// zdd-owed-items, P1): the lease, a fence or the attempt can change during

+// the bucket write, and the exec follows the second answer. A spawn with a

+// gate also honours its context there: one that has ended starts nothing.

+// With neither (every live-link spawn, the flag off) it is a no-op, so those

+// spawns are exactly as they were.

func runExecGate(ctx context.Context) error {

if err := askExecGate(ctx); err != nil {

return err

}

- return runPendingRecord(ctx)

+ wrote, err := runPendingRecord(ctx)

+ if err != nil {

+ return err

+ }

+ if wrote {

+ return askExecGate(ctx)

+ }

+ return nil

}

// RunExecGate is runExecGate for a Backend implemented outside this

@@ -120,7 +128,8 @@ func askExecGate(ctx context.Context) error {

// worker's) before-start hook for the cgroup

// manager (final review F1): launchHook (the test seam), then ctx's exec

// gate, asked immediately before each start attempt, after the cgroup's

-// preparation, and last ctx's pending wake record (runPendingRecord). Nil

+// preparation, then ctx's pending wake record (runPendingRecord), and the

+// gate again once that write has run. Nil

// when ctx carries no gate and no pending record and no test seam is set, so

// a live-link spawn hands the manager no hook and starts exactly as it did.

func (b *ExecBackend) launchGate(ctx context.Context, host string) func() error {

@@ -129,7 +138,8 @@ func (b *ExecBackend) launchGate(ctx context.Context, host string) func() error

// launchGateWith is launchGate with live, a live-link spawn's LiveExecGate

// check for the build it resolved (Task 9), asked after ctx's exec gate at

-// the same point, and before the pending record. A nil live is launchGate

+// the same point, before the pending record and again after its write. A

+// nil live is launchGate

// exactly, so a spawn on a box with no LiveExecGate hands the manager the

// hook it always did.

func (b *ExecBackend) launchGateWith(ctx context.Context, host string, live func() error) func() error {

@@ -148,7 +158,19 @@ func (b *ExecBackend) launchGateWith(ctx context.Context, host string, live func

return err

}

- return runPendingRecord(ctx)

+ wrote, err := runPendingRecord(ctx)

+ if err != nil || !wrote {

+ return err

+ }

+ // The write took time: the gate and the live check are asked again

+ // immediately before the exec (Astra review of zdd-owed-items, P1).

+ if err := askExecGate(ctx); err != nil {

+ return err

+ }

+ if live != nil {

+ return live()

+ }

+ return nil

}

diff --git a/internal/activator/wakeidentity_test.go b/internal/activator/wakeidentity_test.go

index 063f82b90..0b220a689 100644

--- a/internal/activator/wakeidentity_test.go

+++ b/internal/activator/wakeidentity_test.go

@@ -153,7 +153,7 @@ func (b *fakeBackend) StartGenExe(ctx context.Context, rt registry.Route, n uint

}

func (b *fakeBackend) SeedExe(ctx context.Context, rt registry.Route, id release.StagedIdentity) error {

- if err := runPendingRecord(ctx); err != nil {

+ if _, err := runPendingRecord(ctx); err != nil {

return err

}

b.mu.Lock()

diff --git a/internal/activator/wakerecord_recheck_test.go b/internal/activator/wakerecord_recheck_test.go

new file mode 100644

index 000000000..a221e7c25

--- /dev/null

+++ b/internal/activator/wakerecord_recheck_test.go

@@ -0,0 +1,168 @@

+//go:build linux

+package activator

+import (

+ "context"

+ "errors"

+ "os"

+ "path/filepath"

+ "slices"

+ "sync/atomic"

+ "testing"

+ "github.com/carlosframework/platform/internal/instancectl"

+ "github.com/carlosframework/platform/internal/registry"

+// Astra's review of zdd-owed-items. P1: the pending record is a bucket

+// write, and the lease, a fence or the attempt can change while it runs, so

+// the exec is decided again after it. P2: a nested seed writes its

+// obligation before its record, so an obligation that fails records

+// nothing.

+// P1, the exec gate: a gate that passes before the write and refuses after

+// it (the attempt superseded during the PUT) starts nothing, at both seams

+// that carry a gate.

+func TestAnExecGateRefusingDuringTheRecordStartsNothing(t *testing.T) {

+ for _, seam := range []string{"runExecGate", "launchGate"} {

+ t.Run(seam, func(t *testing.T) {

+ var superseded atomic.Bool

+ superseding := errors.New("attempt superseded")

+ gate := func() error {

+ if superseded.Load() {

+ return superseding

+ }

+ return nil

+ }

+ writes := 0

+ write := func() error { writes++; superseded.Store(true); return nil }

+ ctx := withPendingRecord(withExecGate(context.Background(), gate), "v1", write, nil)

+ var err error

+ if seam == "runExecGate" {

+ err = runExecGate(ctx)

+ } else {

+ err = (&ExecBackend{}).launchGate(ctx, "one.example.com")()

+ }

+ if !errors.Is(err, superseding) {

+ t.Fatalf("%s = %v, want the gate's refusal after the record", seam, err)

+ }

+ if writes != 1 {

+ t.Fatalf("%d writes, want 1", writes)

+ }

+ })

+ }

+// P1, the fence: a maintenance fence raised while a cold wake's record is

+// being written starts nothing, at the seed and at the spawn.

+func TestAFenceRaisedDuringTheRecordStartsNothing(t *testing.T) {

+ for _, seeded := range []bool{false, true} {

+ name := "spawn"

+ if seeded {

+ name = "seed"

+ }

+ t.Run(name, func(t *testing.T) {

+ a, b := identityActivator(t, "v1")

+ b.restoreFresh = seeded

+ rt := identityRoute(t, "one.example.com")

+ if seeded {

+ rt.Seed = `["seed"]`

+ }

+ l := &execLog{}

+ a.BeforeExec = func(ctx context.Context, host, build string) error {

+ a.mu.Lock()

+ if a.fences == nil {

+ a.fences = map[string]uint64{}

+ }

+ a.fences[host] = 7 // a maintenance fence lands mid-write

+ a.mu.Unlock()

+ return l.hook(ctx, host, build)

+ }

+ _, err := a.beginRequest(context.Background(), rt)

+ if !errors.Is(err, ErrFenced) {

+ t.Fatalf("wake = %v, want ErrFenced", err)

+ }

+ if b.startCount() != 0 || b.seedCount() != 0 {

+ t.Fatalf("started %d, seeded %d after a fence raised during the record", b.startCount(), b.seedCount())

+ }

+ if calls, _, _ := l.seen(); !slices.Equal(calls, []string{"one.example.com@v1"}) {

+ t.Fatalf("records %v, want the one write that ran", calls)

+ }

+ })

+ }

+// P2: a nested seed whose obligation cannot be written records nothing and

+// asks root for nothing; a nested seed whose record fails discharges the

+// obligation it wrote, keeps the database, and asks root for nothing.

+func TestANestedSeedWritesItsObligationBeforeItsRecord(t *testing.T) {

+ seedOnce := func(t *testing.T, plant func(rt registry.Route), fail bool) (*fakeControl, registry.Route, []string, error) {

+ t.Helper()

+ ctl := &fakeControl{seed: instancectl.SeedResponse{InvocationID: "s1", Result: "success", Status: "0", Code: "exited"}}

+ b, rt := nestedBackend(t, ctl, &fakeUnit{})

+ b.Control = identityControl{ctl}

+ rt.Seed = `["seed"]`

+ stageVersion(t, b, rt.Host, "vA", "exit 0\n")

+ id, err := b.LiveIdentity(context.Background(), rt)

+ if err != nil {

+ t.Fatal(err)

+ }

+ if err := os.MkdirAll(filepath.Dir(rt.DBPath), 0o755); err != nil {

+ t.Fatal(err)

+ }

+ plant(rt)

+ var records []string

+ write := func() error {

+ records = append(records, id.Version)

+ if fail {

+ return errBuildStateUnwritten

+ }

+ return nil

+ }

+ ctx := withPendingRecord(context.Background(), id.Version, write, nil)

+ err = b.SeedExe(ctx, rt, id)

+ return ctl, rt, records, err

+ }

+ t.Run("the obligation fails", func(t *testing.T) {

+ ctl, _, recs, err := seedOnce(t, func(rt registry.Route) {

+ // An earlier seed's unanswered obligation refuses this one.

+ if err := os.WriteFile(rt.DBPath+seedUnprovenSuffix, []byte("{}"), 0o600); err != nil {

+ t.Fatal(err)

+ }

+ }, false)

+ if !errors.Is(err, ErrSeedUnproven) {

+ t.Fatalf("SeedExe = %v, want the obligation's refusal", err)

+ }

+ if len(recs) != 0 {

+ t.Fatalf("records %v, want none: no seed ran", recs)

+ }

+ if ctl.joined() != "" {

+ t.Fatalf("root was asked:\n%s", ctl.joined())

+ }

+ })

+ t.Run("the record fails", func(t *testing.T) {

+ ctl, rt, recs, err := seedOnce(t, func(rt registry.Route) {

+ if err := os.WriteFile(rt.DBPath, []byte("data"), 0o600); err != nil {

+ t.Fatal(err)

+ }

+ }, true)

+ if !errors.Is(err, errBuildStateUnwritten) {

+ t.Fatalf("SeedExe = %v, want the record's refusal", err)

+ }

+ if len(recs) != 1 {

+ t.Fatalf("records %v, want one attempt", recs)

+ }

+ if _, err := os.Stat(rt.DBPath + seedUnprovenSuffix); !errors.Is(err, os.ErrNotExist) {

+ t.Fatalf("the obligation stayed after a refused record: %v", err)

+ }

+ if raw, err := os.ReadFile(rt.DBPath); err != nil || string(raw) != "data" {

+ t.Fatalf("the database was touched: %q, %v", raw, err)

+ }

+ if ctl.joined() != "" {

+ t.Fatalf("root was asked:\n%s", ctl.joined())

+ }

+ })