amadan
zdd-readiness: versioned policy fills, bounded store reads, acks off the scheduler's path
Policy fills are versioned per app and a failed read backs off policyRetryAfter (5 s); the policy, pointer and demand reads run under readBound (5 s), failedUnder under the wake's or the attempt's context. Config acks are numbered under Manager.mu at the transition that releases them and written by one drain goroutine (Manager.acks): newest per host wins, submission never blocks, each write bounded by 10 s; configAckFunc is unchanged for flag off. Manager.Close stops admission, joins in-flight executions and OnGen calls, writes pending acks and stops the writer; the agent calls it after svc.Shutdown. The shared pointer pass requeues a cleared build at once through onCleared (Task 10 review minor 1).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
15 files changed,
+822
−82
internal/host/agent.go+23 −0internal/host/rollout.go+1 −1internal/host/rollout_test.go+4 −1internal/rollout/acks.go+163 −0internal/rollout/config.go+7 −6internal/rollout/config_test.go+10 −9internal/rollout/fake_test.go+8 −0internal/rollout/manager.go+153 −32internal/rollout/manager_test.go+259 −7internal/rollout/policy.go+65 −18internal/rollout/policy_test.go+98 −1internal/rollout/scheduler.go+16 −2internal/rollout/wake_token_test.go+3 −1internal/rollout/wakeplan.go+7 −2internal/rollout/wakeplan_test.go+5 −2
diff --git a/internal/host/agent.go b/internal/host/agent.go| index f4916d876..5421c60f4 100644 |
| --- a/internal/host/agent.go |
| +++ b/internal/host/agent.go |
| @@ -730,6 +730,14 @@ func runAgent(args []string, buildClaims ClaimFactory) { |
| ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) |
| defer cancel() |
| svc.Shutdown(ctx) |
| + // Then the scheduler, under the same budget (spec §8.6, Task 11): an |
| + // attempt whose swap Shutdown just cancelled finishes its done(), its |
| + // outcome and the end of its intent on disk, and so does a wake event |
| + // under way, before this process exits. A planned restart mid-swap so |
| + // leaves nothing for the next agent to reconcile. |
| + if m := armedRollout(); m != nil { |
| + m.Close(ctx) |
| + } |
| // The agent socket's own defer (above, at its construction) stops it |
| // once this function returns — after Shutdown, per the LIFO ordering |
| // explained there. That gap is real, not zero-width: svc.Shutdown |
| @@ -1173,6 +1181,21 @@ func configAckFunc(ctx context.Context, store release.Store, logPrefix string) f |
| } |
| } |
| +// configAckWriter is configAckFunc for the armed scheduler's ack writer |
| +// (rollout.Manager.DeferAck): the same stamp and write, under the context |
| +// the writer hands each write, which bounds it (Task 11, Astra R1 Flag |
| +// triage T8). configAckFunc itself stays as it is, unbounded, for flag-off |
| +// sync and the root pass (plan review #11). |
| +func configAckWriter(store release.Store, logPrefix string) func(context.Context, appconfig.ConfigAck) { |
| + box := BoxLabel() |
| + return func(ctx context.Context, a appconfig.ConfigAck) { |
| + a.Box = box |
| + if err := appconfig.WriteConfigAck(ctx, store, a); err != nil { |
| + log.Printf("%s: config ack %s: %v", logPrefix, a.Host, err) |
| + } |
| + } |
| +} |
| + |
| // reconcileMu serializes the two provisioning loops' Reconcile passes |
| // against each other, the same shape as configSyncMu above. They can |
| // genuinely overlap — instancesLoop's 5-minute pass mid-flight when a |
diff --git a/internal/host/rollout.go b/internal/host/rollout.go| index 56130e037..634eccd3d 100644 |
| --- a/internal/host/rollout.go |
| +++ b/internal/host/rollout.go |
| @@ -435,7 +435,7 @@ func rolloutConfigCallbacks(ctx context.Context, store release.Store, legs confi |
| return appconfig.Callbacks{ |
| Restart: func(string) error { return nil }, |
| Delivered: m.ConfigDelivered, |
| - Ack: m.DeferAck(configAckFunc(ctx, store, "agent")), |
| + Ack: m.DeferAck(configAckWriter(store, "agent")), |
| Managed: rollout.Managed, |
| Unmanaged: legs.restart, |
| }, true |
diff --git a/internal/host/rollout_test.go b/internal/host/rollout_test.go| index 96cdaa83a..1157f55b2 100644 |
| --- a/internal/host/rollout_test.go |
| +++ b/internal/host/rollout_test.go |
| @@ -783,10 +783,13 @@ func TestArmedConfigSyncSplitsManagedFromUnmanaged(t *testing.T) { |
| report := cb.Delivered |
| cb.Delivered = func(h, rev string) { delivered = append(delivered, h); report(h, rev) } |
| acks := map[string]appconfig.ConfigAck{} |
| - cb.Ack = m.DeferAck(func(a appconfig.ConfigAck) { acks[a.Host] = a }) |
| + cb.Ack = m.DeferAck(func(_ context.Context, a appconfig.ConfigAck) { acks[a.Host] = a }) |
| if err := appconfig.SyncExecWithCallbacks(ctx, reg, store, priv, t.TempDir(), "", cb); err != nil { |
| t.Fatal(err) |
| } |
| + // The manager's writer writes the acks (Task 11); Close writes what is |
| + // still pending before it returns, so the map is complete after it. |
| + m.Close(ctx) |
| slices.Sort(delivered) |
| if want := []string{bot.Host, managed.Host}; !slices.Equal(delivered, want) { |
| t.Fatalf("delivered %v, want %v: the managed sidecar's revision must reach the scheduler, the unmanaged route's must not", delivered, want) |
diff --git a/internal/rollout/acks.go b/internal/rollout/acks.go| new file mode 100644 |
| index 000000000..8a4ee05aa |
| --- /dev/null |
| +++ b/internal/rollout/acks.go |
| @@ -0,0 +1,163 @@ |
| +package rollout |
| + |
| +import ( |
| + "context" |
| + "sync" |
| + "time" |
| +) |
| + |
| +// ackWriteBound bounds each config acknowledgement the armed scheduler |
| +// writes (Task 11 rule 3, Astra R1 Flag triage T8): a store that hangs holds |
| +// up the next ack this long, never the scheduler. Flag off, config sync's |
| +// own writer (host.configAckFunc) is unchanged (plan review #11). |
| +const ackWriteBound = 10 * time.Second |
| + |
| +// ackJob is one acknowledgement for the writer: seq is its place in the |
| +// order of the config transitions that released it, taken under Manager.mu |
| +// in the same critical section (Manager.ackLocked), and write is the store |
| +// write, handed the bounded context. |
| +type ackJob struct { |
| + seq uint64 |
| + write func(context.Context) |
| +} |
| + |
| +// ackWriter is Manager.acks: the armed scheduler's acknowledgement writes, |
| +// off the paths that release them. A config activation releases its held ack |
| +// in done() before the attempt's slot frees, and in the scheduler loop for a |
| +// sleeper (configSlept); neither may wait on the store, so they submit here |
| +// and one drain goroutine writes, with no lock held while it writes (Task 11 |
| +// rule 3). |
| +// |
| +// The order is fixed at the transition, not at the write (plan review #10): |
| +// a host's acks only ever replace each other, so the writer keeps the newest |
| +// pending one per host, by sequence, and drops an older one, whether it |
| +// arrives after a newer one is pending or after a newer one was written. |
| +// An ack written after a newer one would name a revision that is no longer |
| +// the delivered one (spec §8.1). Hosts are written in the order their |
| +// pending ack was first queued. Submission never blocks. |
| +type ackWriter struct { |
| + ctx context.Context |
| + logf func(format string, args ...any) |
| + kick chan struct{} |
| + done chan struct{} // closed when the drain goroutine has returned |
| + |
| + mu sync.Mutex |
| + pending map[string]ackJob |
| + order []string // hosts with a pending ack, first queued first |
| + newest map[string]uint64 // per host, the newest sequence accepted |
| + closing bool // Close: drain what is pending, then return |
| + stop bool // return after the write in progress |
| + writing bool |
| + // onSubmit, when set, runs at every accepted submission, with no lock |
| + // held: a test's window into the moment an ack is released. |
| + onSubmit func(host string) |
| +} |
| + |
| +// newAckWriter starts the drain goroutine. It runs until ctx ends or close |
| +// returns; a write in progress when ctx ends is finished (its own context is |
| +// ctx's, so the store call returns as soon as it notices) and nothing more is |
| +// written. |
| +func newAckWriter(ctx context.Context, logf func(format string, args ...any)) *ackWriter { |
| + w := &ackWriter{ctx: ctx, logf: logf, kick: make(chan struct{}, 1), done: make(chan struct{}), |
| + pending: map[string]ackJob{}, newest: map[string]uint64{}} |
| + go w.drain() |
| + return w |
| +} |
| + |
| +// submit queues j as host's ack unless an ack with a newer sequence was |
| +// already queued or written for host. It never blocks. |
| +func (w *ackWriter) submit(host string, j ackJob) { |
| + w.mu.Lock() |
| + if w.stop || j.seq <= w.newest[host] { |
| + w.mu.Unlock() |
| + return |
| + } |
| + w.newest[host] = j.seq |
| + if _, queued := w.pending[host]; !queued { |
| + w.order = append(w.order, host) |
| + } |
| + w.pending[host] = j |
| + hook := w.onSubmit |
| + w.mu.Unlock() |
| + if hook != nil { |
| + hook(host) |
| + } |
| + select { |
| + case w.kick <- struct{}{}: |
| + default: |
| + } |
| +} |
| + |
| +// next takes the first queued host's ack. ok is false when there is none, |
| +// or the writer is to stop. |
| +func (w *ackWriter) next() (host string, j ackJob, ok bool) { |
| + w.mu.Lock() |
| + defer w.mu.Unlock() |
| + if w.stop || len(w.order) == 0 || w.ctx.Err() != nil { |
| + w.writing = false |
| + return "", ackJob{}, false |
| + } |
| + host, w.order = w.order[0], w.order[1:] |
| + j = w.pending[host] |
| + delete(w.pending, host) |
| + w.writing = true |
| + return host, j, true |
| +} |
| + |
| +func (w *ackWriter) drain() { |
| + defer close(w.done) |
| + for { |
| + for { |
| + _, j, ok := w.next() |
| + if !ok { |
| + break |
| + } |
| + ctx, cancel := context.WithTimeout(w.ctx, ackWriteBound) |
| + j.write(ctx) |
| + cancel() |
| + } |
| + w.mu.Lock() |
| + finished := w.stop || (w.closing && len(w.order) == 0) |
| + w.mu.Unlock() |
| + if finished { |
| + return |
| + } |
| + select { |
| + case <-w.ctx.Done(): |
| + return |
| + case <-w.kick: |
| + } |
| + } |
| +} |
| + |
| +// idle reports whether nothing is pending or being written. |
| +func (w *ackWriter) idle() bool { |
| + w.mu.Lock() |
| + defer w.mu.Unlock() |
| + return len(w.order) == 0 && !w.writing |
| +} |
| + |
| +// close writes what is pending and stops the writer, waiting until ctx |
| +// ends. Once ctx ends the writer stops after the write in progress, and |
| +// close returns without waiting for it: what is still pending then is not |
| +// written. An ack submitted after close is dropped. |
| +func (w *ackWriter) close(ctx context.Context) { |
| + w.mu.Lock() |
| + w.closing = true |
| + w.mu.Unlock() |
| + select { |
| + case w.kick <- struct{}{}: |
| + default: |
| + } |
| + select { |
| + case <-w.done: |
| + case <-ctx.Done(): |
| + } |
| + w.mu.Lock() |
| + w.stop = true |
| + left := len(w.order) |
| + w.mu.Unlock() |
| + if left > 0 { |
| + w.logf("rollout: %d config acknowledgement(s) not written before the agent stopped", left) |
| + } |
| +} |
diff --git a/internal/rollout/config.go b/internal/rollout/config.go| index cd3c14d32..fe150288b 100644 |
| --- a/internal/rollout/config.go |
| +++ b/internal/rollout/config.go |
| @@ -1,6 +1,7 @@ |
| package rollout |
| import ( |
| + "context" |
| "time" |
| "github.com/carlosframework/platform/internal/appconfig" |
| @@ -21,7 +22,7 @@ type configTrack struct { |
| ack *appconfig.ConfigAck |
| ackRev string |
| - ackWrite func(appconfig.ConfigAck) |
| + ackWrite func(context.Context, appconfig.ConfigAck) |
| } |
| // configBackoff is the wait after the nth consecutive failed config attempt: |
| @@ -40,17 +41,17 @@ func configBackoff(n int) time.Duration { |
| // restart) acknowledges at once, as before. So does a sleeping instance, |
| // with today's hibernating acknowledgement: the config landed and nothing |
| // woke to run it (ConfigAck.Hibernating, "delivered on wake"). |
| -func (t *configTrack) deferAck(a appconfig.ConfigAck, write func(appconfig.ConfigAck)) func() { |
| +func (t *configTrack) deferAck(a appconfig.ConfigAck, write func(context.Context, appconfig.ConfigAck)) func(context.Context) { |
| // A newer delivery supersedes any ack still held for an older one, whether |
| // or not this one waits: releasing the old ack later would write it after |
| // the newer ack, with a revision that is no longer delivered (spec §8.1). |
| t.ack, t.ackWrite, t.ackRev = nil, nil, "" |
| if t.sleeping { |
| a.Hibernating, a.Restarted = true, false |
| - return func() { write(a) } |
| + return func(ctx context.Context) { write(ctx, a) } |
| } |
| if t.delivered == "" || t.active == t.delivered { |
| - return func() { write(a) } |
| + return func(ctx context.Context) { write(ctx, a) } |
| } |
| t.ack, t.ackRev, t.ackWrite = &a, t.delivered, write |
| return nil |
| @@ -58,7 +59,7 @@ func (t *configTrack) deferAck(a appconfig.ConfigAck, write func(appconfig.Confi |
| // activate records rev as active and returns the held ack when it was |
| // waiting for exactly rev and rev is still the delivered revision. |
| -func (t *configTrack) activate(rev string) (func(), bool) { |
| +func (t *configTrack) activate(rev string) (func(context.Context), bool) { |
| if rev == "" { |
| return nil, false |
| } |
| @@ -71,5 +72,5 @@ func (t *configTrack) activate(rev string) (func(), bool) { |
| } |
| a, write := *t.ack, t.ackWrite |
| t.ack, t.ackWrite, t.ackRev = nil, nil, "" |
| - return func() { write(a) }, true |
| + return func(ctx context.Context) { write(ctx, a) }, true |
| } |
diff --git a/internal/rollout/config_test.go b/internal/rollout/config_test.go| index 62c1eb22a..44bdd8ea2 100644 |
| --- a/internal/rollout/config_test.go |
| +++ b/internal/rollout/config_test.go |
| @@ -1,6 +1,7 @@ |
| package rollout |
| import ( |
| + "context" |
| "testing" |
| "time" |
| @@ -18,7 +19,7 @@ func TestConfigBackoffRunsFromOneMinuteToThirty(t *testing.T) { |
| func TestAnAckWaitsForItsRevisionToBeActive(t *testing.T) { |
| var wrote []string |
| - write := func(a appconfig.ConfigAck) { wrote = append(wrote, a.Host) } |
| + write := func(_ context.Context, a appconfig.ConfigAck) { wrote = append(wrote, a.Host) } |
| tr := &configTrack{delivered: "c2", active: "c1"} |
| if now := tr.deferAck(appconfig.ConfigAck{Host: "a"}, write); now != nil { |
| t.Fatal("acknowledged a delivered revision that is not active") |
| @@ -30,7 +31,7 @@ func TestAnAckWaitsForItsRevisionToBeActive(t *testing.T) { |
| if !ok { |
| t.Fatal("the ack for c2 did not come back when c2 became active") |
| } |
| - run() |
| + run(context.Background()) |
| if len(wrote) != 1 { |
| t.Fatalf("wrote %v", wrote) |
| } |
| @@ -48,11 +49,11 @@ func TestAnAckWaitsForItsRevisionToBeActive(t *testing.T) { |
| func TestASleepersAckGoesOutAtDeliveryAsHibernating(t *testing.T) { |
| var got []appconfig.ConfigAck |
| tr := &configTrack{delivered: "c2", active: "c2", sleeping: true} |
| - run := tr.deferAck(appconfig.ConfigAck{Host: "a", Restarted: true}, func(a appconfig.ConfigAck) { got = append(got, a) }) |
| + run := tr.deferAck(appconfig.ConfigAck{Host: "a", Restarted: true}, func(_ context.Context, a appconfig.ConfigAck) { got = append(got, a) }) |
| if run == nil { |
| t.Fatal("a sleeper's ack waited for a wake") |
| } |
| - run() |
| + run(context.Background()) |
| if len(got) != 1 || !got[0].Hibernating || got[0].Restarted { |
| t.Fatalf("ack = %+v, want Hibernating and not Restarted", got) |
| } |
| @@ -64,7 +65,7 @@ func TestASleepersAckGoesOutAtDeliveryAsHibernating(t *testing.T) { |
| func TestANewerAckSupersedesTheHeldOne(t *testing.T) { |
| t.Run("revert to the active revision", func(t *testing.T) { |
| var wrote []string |
| - write := func(a appconfig.ConfigAck) { wrote = append(wrote, a.Host) } |
| + write := func(_ context.Context, a appconfig.ConfigAck) { wrote = append(wrote, a.Host) } |
| tr := &configTrack{delivered: "c2", active: "c1"} |
| if now := tr.deferAck(appconfig.ConfigAck{Host: "held-c2"}, write); now != nil { |
| t.Fatal("c2's ack was not held while c1 is active") |
| @@ -75,7 +76,7 @@ func TestANewerAckSupersedesTheHeldOne(t *testing.T) { |
| if now == nil { |
| t.Fatal("c1's ack waited though c1 is active") |
| } |
| - now() |
| + now(context.Background()) |
| // The in-flight generation finishes on c2. |
| if run, ok := tr.activate("c2"); run != nil || ok { |
| t.Fatal("the stale c2 ack was released after a newer ack went out") |
| @@ -86,7 +87,7 @@ func TestANewerAckSupersedesTheHeldOne(t *testing.T) { |
| }) |
| t.Run("sleep then hibernating ack", func(t *testing.T) { |
| var wrote []string |
| - write := func(a appconfig.ConfigAck) { wrote = append(wrote, a.Host) } |
| + write := func(_ context.Context, a appconfig.ConfigAck) { wrote = append(wrote, a.Host) } |
| tr := &configTrack{delivered: "c2", active: "c1"} |
| if now := tr.deferAck(appconfig.ConfigAck{Host: "held-c2"}, write); now != nil { |
| t.Fatal("c2's ack was not held while c1 is active") |
| @@ -97,7 +98,7 @@ func TestANewerAckSupersedesTheHeldOne(t *testing.T) { |
| if now == nil { |
| t.Fatal("a sleeper's ack waited") |
| } |
| - now() |
| + now(context.Background()) |
| if run, ok := tr.activate("c2"); run != nil || ok { |
| t.Fatal("the stale c2 ack was released after the Hibernating ack went out") |
| } |
| @@ -111,7 +112,7 @@ func TestANewerAckSupersedesTheHeldOne(t *testing.T) { |
| // when delivered moved on without another ack arriving. |
| func TestAHeldAckIsNotReleasedForARevisionNoLongerDelivered(t *testing.T) { |
| tr := &configTrack{delivered: "c2", active: "c1"} |
| - tr.deferAck(appconfig.ConfigAck{Host: "a"}, func(appconfig.ConfigAck) { t.Fatal("wrote a stale ack") }) |
| + tr.deferAck(appconfig.ConfigAck{Host: "a"}, func(context.Context, appconfig.ConfigAck) { t.Fatal("wrote a stale ack") }) |
| tr.delivered = "c3" |
| if run, ok := tr.activate("c2"); run != nil || ok { |
| t.Fatal("released an ack for c2 while c3 is delivered") |
diff --git a/internal/rollout/fake_test.go b/internal/rollout/fake_test.go| index 2a4173ab5..62b93f534 100644 |
| --- a/internal/rollout/fake_test.go |
| +++ b/internal/rollout/fake_test.go |
| @@ -465,6 +465,14 @@ func (h *harness) servingAll(build string, hosts ...string) bool { |
| return true |
| } |
| +// flushAcks waits until the manager's ack writer has written every ack |
| +// handed to it so far. Acks are written off the path that releases them |
| +// (Task 11), so a test reads what was written only after this. |
| +func (h *harness) flushAcks() { |
| + h.t.Helper() |
| + waitFor(h.t, "the ack writer to finish", h.m.acks.idle) |
| +} |
| + |
| func waitFor(t *testing.T, what string, cond func() bool) { |
| t.Helper() |
| deadline := time.Now().Add(5 * time.Second) |
diff --git a/internal/rollout/manager.go b/internal/rollout/manager.go| index 5c76e01e1..1bc74b5f6 100644 |
| --- a/internal/rollout/manager.go |
| +++ b/internal/rollout/manager.go |
| @@ -160,10 +160,26 @@ type Manager struct { |
| // dispatch.ExecID). |
| execCount uint64 |
| + // ackSeq numbers the config acks the tracks hand back, under mu, in the |
| + // critical section of the transition that releases each one: the |
| + // writer's order (ackLocked, plan review #10). |
| + ackSeq uint64 |
| + // acks writes those acks, off the scheduler's slot and loop (Task 11). |
| + acks *ackWriter |
| + |
| // goodMu serialises every serving- and good-build write with its |
| // acknowledgement check (served, publishServing, writeGood), so an older |
| // generation's write can never land after a newer one's (spec §8.5). |
| goodMu sync.Mutex |
| + |
| + // The shutdown join (Close, spec §8.6). joinMu is a leaf: nothing else |
| + // is taken while it is held. inflight counts the attempts' executions and |
| + // the OnGen calls under way; closed stops admission; joined, when set, |
| + // is closed once inflight is zero. |
| + joinMu sync.Mutex |
| + closed bool |
| + inflight int |
| + joined chan struct{} |
| } |
| // NewManager builds the manager. Its schedulers run until ctx ends. |
| @@ -178,10 +194,75 @@ func NewManager(ctx context.Context, d Deps) *Manager { |
| apps: map[appKey]*scheduler{}, hostApp: map[string]appKey{}, cfg: map[string]*configTrack{}, |
| retried: map[string]string{}, pointerRead: map[appKey]time.Time{}, nudging: map[string]bool{}, observing: map[string]uint64{}, |
| planned: map[string]plannedWake{}, wakeTok: map[string]uint64{}, wakeAt: map[string][]wakeStart{}} |
| - m.policies = &policyCache{load: d.Policy, ttl: 30 * time.Second, now: d.Now, m: map[appKey]policyEntry{}} |
| + m.policies = &policyCache{load: d.Policy, ttl: 30 * time.Second, now: d.Now, m: map[appKey]*policyEntry{}} |
| + m.acks = newAckWriter(ctx, d.Logf) |
| return m |
| } |
| +// Close is the scheduler's part of a graceful stop (spec §8.6: shutdown |
| +// cancels and joins the whole operation set; Task 5 review). The agent calls |
| +// it after the activator's Shutdown, which has stopped every entry and so |
| +// cancelled any swap in flight. It stops admitting attempts, then waits, |
| +// until ctx ends, for every attempt's execution to finish its done() (the |
| +// outcome's book writes and the intent's End) and for every OnGen call |
| +// under way to return, and then writes the acks still pending and stops the |
| +// ack writer. A planned restart in the middle of a swap so leaves the |
| +// attempt's outcome on disk and no intent behind, and the next agent |
| +// reconciles nothing. A Close that runs out of ctx returns anyway: what was |
| +// not finished is what OpenFailureBook reconciles. |
| +func (m *Manager) Close(ctx context.Context) { |
| + m.joinMu.Lock() |
| + m.closed = true |
| + var wait chan struct{} |
| + if m.inflight > 0 { |
| + if m.joined == nil { |
| + m.joined = make(chan struct{}) |
| + } |
| + wait = m.joined |
| + } |
| + n := m.inflight |
| + m.joinMu.Unlock() |
| + if wait != nil { |
| + select { |
| + case <-wait: |
| + case <-ctx.Done(): |
| + m.d.Logf("rollout: stopping with %d attempt(s) or wake event(s) still running: %v", n, ctx.Err()) |
| + } |
| + } |
| + m.acks.close(ctx) |
| +} |
| + |
| +// enter counts one operation Close joins. An admission (admit true) is |
| +// refused once Close has begun; an OnGen call is always counted, so one that |
| +// arrives while Close waits is waited for too. |
| +func (m *Manager) enter(admit bool) bool { |
| + m.joinMu.Lock() |
| + defer m.joinMu.Unlock() |
| + if admit && m.closed { |
| + return false |
| + } |
| + m.inflight++ |
| + return true |
| +} |
| + |
| +// leave ends an operation enter counted. |
| +func (m *Manager) leave() { |
| + m.joinMu.Lock() |
| + defer m.joinMu.Unlock() |
| + m.inflight-- |
| + if m.inflight == 0 && m.joined != nil { |
| + close(m.joined) |
| + m.joined = nil |
| + } |
| +} |
| + |
| +// closing reports whether Close has begun. |
| +func (m *Manager) closing() bool { |
| + m.joinMu.Lock() |
| + defer m.joinMu.Unlock() |
| + return m.closed |
| +} |
| + |
| // changed tells OnChange a transition happened. Every caller has released |
| // Manager.mu and any scheduler.mu first, so the callback may read Status. |
| func (m *Manager) changed() { |
| @@ -491,10 +572,9 @@ func (m *Manager) applyPointers(routes []registry.Route, keys []appKey) { |
| if !due { |
| continue |
| } |
| - cleared, unmarked := w.apply(m.ctx, m.d.Book, m.d.Pointer, routes, k, m.d.Logf) |
| - for _, build := range cleared { |
| - m.retryBuild(k, build, "") |
| - } |
| + // Each cleared record's build is requeued as a retry at once, before |
| + // the app's next channel is read (Task 10 review minor 1). |
| + _, unmarked := w.apply(m.ctx, m.d.Book, m.d.Pointer, routes, k, m.d.Logf, func(build string) { m.retryBuild(k, build, "") }) |
| if unmarked { |
| m.changed() |
| } |
| @@ -531,11 +611,15 @@ func pointerWorkFor(book *FailureBook, routes []registry.Route, k appKey) pointe |
| // pointer names with a newer PromotedAt is cleared (ClearOnPointer), and so |
| // is the marker of a host whose channel's pointer names its installed good |
| // build with a PromotedAt after the marker's At (ClearHoldRanOnPointer). A |
| -// read error is logged and that channel skipped. It reports the builds whose |
| -// records it cleared and whether it cleared a marker; what follows from |
| -// either is the caller's. |
| +// read error is logged and that channel skipped; each read is bounded by |
| +// readBound (Task 11 rule 2). onCleared is called with each cleared record's |
| +// build as soon as the book has cleared it, before the next record or |
| +// channel, with no lock held: the armed pass requeues the build there, so |
| +// the retry is not left waiting on the rest of the app's reads (Task 10 |
| +// review minor 1). It reports how many records it cleared and whether it |
| +// cleared a marker; what follows from a marker is the caller's. |
| func (w pointerWork) apply(ctx context.Context, book *FailureBook, read PointerRead, routes []registry.Route, k appKey, |
| - logf func(format string, args ...any)) (cleared []string, unmarked bool) { |
| + logf func(format string, args ...any), onCleared func(build string)) (cleared int, unmarked bool) { |
| channels := map[string]bool{} |
| for _, rt := range routes { |
| if Managed(rt) && release.RouteAccount(rt) == k.acct && rt.App == k.app { |
| @@ -543,7 +627,9 @@ func (w pointerWork) apply(ctx context.Context, book *FailureBook, read PointerR |
| } |
| } |
| for ch := range channels { |
| - ptr, err := read(ctx, k.acct, k.app, ch) |
| + rctx, cancel := context.WithTimeout(ctx, readBound) |
| + ptr, err := read(rctx, k.acct, k.app, ch) |
| + cancel() |
| if err != nil { |
| logf("rollout: %s/%s channel %s: %v (retry not checked this pass)", k.acct, k.app, ch, err) |
| continue |
| @@ -560,7 +646,8 @@ func (w pointerWork) apply(ctx context.Context, book *FailureBook, read PointerR |
| logf("rollout: %s/%s clear record for %s: %v (cleared in memory)", k.acct, k.app, r.Build, err) |
| } |
| if ok { |
| - cleared = append(cleared, r.Build) |
| + cleared++ |
| + onCleared(r.Build) |
| } |
| } |
| for _, rt := range w.rollback[ch] { |
| @@ -609,11 +696,10 @@ func ApplyPointers(ctx context.Context, book *FailureBook, read PointerRead, rou |
| if w.empty() { |
| continue |
| } |
| - cleared, unmarked := w.apply(ctx, book, read, routes, k, logf) |
| - for _, build := range cleared { |
| + cleared, unmarked := w.apply(ctx, book, read, routes, k, logf, func(build string) { |
| logf("rollout: %s/%s: a signed pointer re-promoted %s, so its failure record is cleared", k.acct, k.app, build) |
| - } |
| - changed = changed || len(cleared) > 0 || unmarked |
| + }) |
| + changed = changed || cleared > 0 || unmarked |
| } |
| return changed |
| } |
| @@ -713,9 +799,15 @@ func (m *Manager) RetryHost(host, build string, epoch uint64) bool { |
| // read, or already names another build, it is zero, and the failure's own |
| // time stands in when the record is written: only a pointer signed after the |
| // failure can then clear the record, which is the safe direction. |
| -func (m *Manager) failedUnder(rt registry.Route, build string) time.Time { |
| +// |
| +// ctx is the caller's: the wake's in planRow, the attempt's in execute |
| +// (Task 11 rule 2, Astra R1 Flag triage T10). A cancelled wake no longer |
| +// waits on the store for its pointer, and no read waits past readBound. |
| +func (m *Manager) failedUnder(ctx context.Context, rt registry.Route, build string) time.Time { |
| if m.d.Pointer != nil { |
| - ptr, err := m.d.Pointer(m.ctx, release.RouteAccount(rt), rt.App, routeChannel(rt)) |
| + ctx, cancel := context.WithTimeout(ctx, readBound) |
| + defer cancel() |
| + ptr, err := m.d.Pointer(ctx, release.RouteAccount(rt), rt.App, routeChannel(rt)) |
| if err == nil && ptr != nil && ptr.Version == build { |
| return ptr.PromotedAt |
| } |
| @@ -1053,15 +1145,37 @@ func (m *Manager) ConfigDelivered(host, rev string) { |
| } |
| // DeferAck wraps config sync's ack writer: a revision is acknowledged only |
| -// once it is active (spec §8.1). |
| -func (m *Manager) DeferAck(write func(appconfig.ConfigAck)) func(appconfig.ConfigAck) { |
| +// once it is active (spec §8.1). Every ack, one written at once included, |
| +// goes through the manager's writer (Manager.acks), numbered here under m.mu |
| +// with the transitions that release held ones: written in line, an ack at |
| +// delivery could land before an older held ack the writer still has, and be |
| +// overwritten by it (plan review #10). write is handed a context bounded by |
| +// ackWriteBound. |
| +func (m *Manager) DeferAck(write func(context.Context, appconfig.ConfigAck)) func(appconfig.ConfigAck) { |
| return func(a appconfig.ConfigAck) { |
| m.mu.Lock() |
| - now := m.track(a.Host).deferAck(a, write) |
| + now := m.ackLocked(m.track(a.Host).deferAck(a, write)) |
| m.mu.Unlock() |
| - if now != nil { |
| - now() |
| - } |
| + m.submitAck(a.Host, now) |
| + } |
| +} |
| + |
| +// ackLocked numbers an ack a config track handed back (nil when none). The |
| +// caller holds m.mu and is in the critical section of the transition that |
| +// released it, so the numbers follow the order of the transitions, which |
| +// the writer keeps per host (plan review #10). |
| +func (m *Manager) ackLocked(write func(context.Context)) ackJob { |
| + if write == nil { |
| + return ackJob{} |
| + } |
| + m.ackSeq++ |
| + return ackJob{seq: m.ackSeq, write: write} |
| +} |
| + |
| +// submitAck hands an ack to the writer; it never blocks. |
| +func (m *Manager) submitAck(host string, j ackJob) { |
| + if j.write != nil { |
| + m.acks.submit(host, j) |
| } |
| } |
| @@ -1109,7 +1223,11 @@ func (m *Manager) stagedDemand(ctx context.Context, rt registry.Route, build, pa |
| if _, err := os.Stat(path); err != nil { |
| return "", "", err |
| } |
| - demand, digest, err = m.d.Demand(ctx, release.RouteAccount(rt), rt.App, build, path) |
| + // Bounded (Task 11 rule 2): a demand read that times out is a read |
| + // error, so the attempt or the wake runs in hold, as below. |
| + rctx, cancel := context.WithTimeout(ctx, readBound) |
| + demand, digest, err = m.d.Demand(rctx, release.RouteAccount(rt), rt.App, build, path) |
| + cancel() |
| if errors.Is(err, release.ErrStagedDigestMismatch) { |
| return "", "", err |
| } |
| @@ -1272,23 +1390,25 @@ func (m *Manager) configActivated(host, rev string) { |
| } |
| m.mu.Lock() |
| write, _ := m.track(host).activate(rev) |
| + ack := m.ackLocked(write) |
| m.mu.Unlock() |
| - m.activationDone(host, rev, write) |
| + m.activationDone(host, rev, ack) |
| } |
| // activationDone is the rest of an activation of rev, after the config |
| // track has recorded it under m.mu: the queue's pending config attempt for |
| -// rev is cleared, and the held ack (write, nil when none) is written. Called |
| -// with no lock held: the ack is I/O. |
| -func (m *Manager) activationDone(host, rev string, write func()) { |
| +// rev is cleared, and the held ack (ack, empty when none), numbered in that |
| +// critical section, is handed to the writer. Called with no lock held. The |
| +// ack is not written here (Task 11 rule 3, Astra R1 Flag triage T8): done() |
| +// calls this before the attempt's slot frees, and configSlept from the |
| +// scheduler loop, and neither may wait on the store. |
| +func (m *Manager) activationDone(host, rev string, ack ackJob) { |
| if s := m.schedulerOf(host); s != nil { |
| s.mu.Lock() |
| s.q.configActive(host, rev) |
| s.mu.Unlock() |
| } |
| - if write != nil { |
| - write() |
| - } |
| + m.submitAck(host, ack) |
| } |
| // configSlept is configActivated for an instance seen asleep with rev |
| @@ -1304,8 +1424,9 @@ func (m *Manager) configSlept(host, rev string) { |
| t.ack.Hibernating, t.ack.Restarted = true, false |
| } |
| write, _ := t.activate(rev) |
| + ack := m.ackLocked(write) |
| m.mu.Unlock() |
| - m.activationDone(host, rev, write) |
| + m.activationDone(host, rev, ack) |
| } |
| // configFailure counts a failed attempt that carried config revision rev |
diff --git a/internal/rollout/manager_test.go b/internal/rollout/manager_test.go| index 045380285..d0337b8fa 100644 |
| --- a/internal/rollout/manager_test.go |
| +++ b/internal/rollout/manager_test.go |
| @@ -1,6 +1,7 @@ |
| package rollout |
| import ( |
| + "context" |
| "errors" |
| "fmt" |
| "strings" |
| @@ -346,7 +347,7 @@ func TestConfigDeliveredMidSwapIsAckedOnlyWhenActive(t *testing.T) { |
| waitFor(t, "the release swap to start", func() bool { _, s, _ := h.act.snapshot(); return len(s) == 1 }) |
| // The ack is written from the scheduler's goroutine: counted atomically. |
| var acks atomic.Int32 |
| - ack := h.m.DeferAck(func(appconfig.ConfigAck) { acks.Add(1) }) |
| + ack := h.m.DeferAck(func(context.Context, appconfig.ConfigAck) { acks.Add(1) }) |
| h.m.ConfigDelivered("a", "sha256:c1") |
| ack(appconfig.ConfigAck{Host: "a"}) |
| if acks.Load() != 0 { |
| @@ -365,20 +366,23 @@ func TestConfigDeliveredMidSwapIsAckedOnlyWhenActive(t *testing.T) { |
| // its slot frees, and with no registry read in front of it. Published after |
| // the slot and the read, a newer config attempt could complete first and be |
| // overwritten (active went back to the older revision), and an unreadable |
| -// row lost the activation, leaving delivered and active apart for good. |
| +// row lost the activation, leaving delivered and active apart for good. The |
| +// held ack is released (handed to the writer, its order fixed) in the same |
| +// step; the write itself is the writer's, after (Task 11). |
| func TestConfigActivationIsPublishedBeforeTheSlotFrees(t *testing.T) { |
| h := newHarness(t, "a") |
| h.pass() |
| - var acks, freed atomic.Int32 |
| - ack := h.m.DeferAck(func(appconfig.ConfigAck) { |
| - acks.Add(1) |
| + var acks, released, freed atomic.Int32 |
| + h.m.acks.onSubmit = func(host string) { |
| + released.Add(1) |
| s := h.m.schedulerOf("a") |
| s.mu.Lock() |
| if hs := s.q.hosts["a"]; hs != nil && hs.running == nil { |
| freed.Add(1) |
| } |
| s.mu.Unlock() |
| - }) |
| + } |
| + ack := h.m.DeferAck(func(context.Context, appconfig.ConfigAck) { acks.Add(1) }) |
| var swapped atomic.Bool |
| h.act.outcome = func(activator.Attempt) activator.SwapOutcome { |
| swapped.Store(true) |
| @@ -403,6 +407,9 @@ func TestConfigActivationIsPublishedBeforeTheSlotFrees(t *testing.T) { |
| } |
| close(gate) |
| waitFor(t, "the config activated and acknowledged", func() bool { return acks.Load() == 1 }) |
| + if released.Load() != 1 { |
| + t.Fatalf("%d acks released, want 1", released.Load()) |
| + } |
| if n := freed.Load(); n != 0 { |
| t.Fatal("the activation was published after the attempt's slot was freed") |
| } |
| @@ -796,7 +803,7 @@ func TestASleepersPendingConfigIsNoAttemptAfterItWakes(t *testing.T) { |
| h.pass() |
| var mu sync.Mutex |
| var acks []appconfig.ConfigAck |
| - ack := h.m.DeferAck(func(a appconfig.ConfigAck) { |
| + ack := h.m.DeferAck(func(_ context.Context, a appconfig.ConfigAck) { |
| mu.Lock() |
| defer mu.Unlock() |
| acks = append(acks, a) |
| @@ -1029,3 +1036,248 @@ func TestReasonWords(t *testing.T) { |
| // configPendingHead is the config-pending event up to its reason: the tests |
| // that wait for it do not know the reason or the backoff. |
| var configPendingHead = textHead(rolloutwords.EventConfigPending, "%s") |
| + |
| +// Task 11 rule 3 (Astra R1 Flag triage T8): the held ack an activation |
| +// releases is handed to the manager's writer, not written by done(). A store |
| +// write that never returns cannot keep the attempt's slot, so the host's |
| +// next attempt is admitted while the writer is still blocked. |
| +func TestABlockedAckWriterDoesNotHoldTheSlot(t *testing.T) { |
| + h := newHarness(t, "a") |
| + h.pass() |
| + h.stage("v2") |
| + entered, block := make(chan struct{}, 4), make(chan struct{}) |
| + t.Cleanup(func() { close(block) }) |
| + ack := h.m.DeferAck(func(context.Context, appconfig.ConfigAck) { |
| + entered <- struct{}{} |
| + <-block |
| + }) |
| + h.m.ConfigDelivered("a", "sha256:c1") |
| + ack(appconfig.ConfigAck{Host: "a"}) |
| + select { |
| + case <-entered: |
| + case <-time.After(5 * time.Second): |
| + t.Fatal("the config attempt's ack was never written") |
| + } |
| + h.install("v2", "a") |
| + h.pass() |
| + waitFor(t, "the release attempt, admitted while the ack writer blocks", func() bool { return h.servingAll("v2", "a") }) |
| +} |
| + |
| +// Task 11 rule 2 (Astra R1 Flag triage T10): the failing pointer a wake's |
| +// plan reads runs under the wake's context. A store read that never answers |
| +// ends when the wake is cancelled, and the plan goes on with no PromotedAt |
| +// (the failure's own time stands in, the safe direction). |
| +func TestABlockedPointerReadEndsWithTheWake(t *testing.T) { |
| + h := newHarness(t, "a") |
| + h.stage("v2") |
| + h.install("v2", "a") |
| + reading := make(chan struct{}) |
| + var once sync.Once |
| + h.m.d.Pointer = func(ctx context.Context, _, _, _ string) (*release.ChannelPointer, error) { |
| + once.Do(func() { close(reading) }) |
| + <-ctx.Done() |
| + return nil, ctx.Err() |
| + } |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + defer cancel() |
| + plans := make(chan activator.WakePlan, 1) |
| + go func() { plans <- h.m.Plan(ctx, h.rows.get("a")) }() |
| + <-reading |
| + cancel() |
| + select { |
| + case p := <-plans: |
| + if !p.FirstWake || !p.FailedUnder.IsZero() { |
| + t.Fatalf("plan %+v: want the first wake of v2 with no PromotedAt", p) |
| + } |
| + case <-time.After(2 * time.Second): |
| + t.Fatal("the plan's pointer read outlived the wake's context") |
| + } |
| +} |
| + |
| +// Task 11 Corrections (Task 5 review; spec §8.6): a planned agent restart |
| +// stops the activator first, which cancels a swap mid-flight, and then joins |
| +// the scheduler. Close returns only once that attempt's done() has written |
| +// its outcome (here the hold-ran marker of the hold build that ran) and |
| +// ended its intent, so the restart reconciles nothing. After Close no |
| +// attempt is admitted. |
| +func TestCloseJoinsAnAttemptCancelledByShutdown(t *testing.T) { |
| + h := newHarness(t, "a") |
| + dir := h.onDisk() |
| + h.swap = "hold" |
| + h.stage("v2", "v3") |
| + gate := h.act.hold("a") |
| + h.act.outcome = func(activator.Attempt) activator.SwapOutcome { |
| + time.Sleep(200 * time.Millisecond) // the swap winding down after its entry stopped |
| + return activator.SwapAbandoned |
| + } |
| + h.act.executed = func(activator.Attempt, activator.SwapOutcome) bool { return true } // the hold build ran |
| + h.install("v2", "a") |
| + h.pass() |
| + waitFor(t, "the hold swap to start", func() bool { _, s, _ := h.act.snapshot(); return len(s) == 1 }) |
| + if n := len(fileOf(t, dir).Intents); n != 1 { |
| + t.Fatalf("%d intents on disk while the swap runs, want 1", n) |
| + } |
| + close(gate) // svc.Shutdown stops the entry under the swap |
| + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) |
| + defer cancel() |
| + h.m.Close(ctx) |
| + f := fileOf(t, dir) |
| + if len(f.Intents) != 0 { |
| + t.Fatalf("intents %+v on disk after Close: a restart would reconcile the attempt as a failure", f.Intents) |
| + } |
| + if mk, ok := f.HoldRan["a"]; !ok || mk.Build != "v2" { |
| + t.Fatalf("hold-ran markers %+v after Close, want a's for v2: the attempt's outcome", f.HoldRan) |
| + } |
| + h.advance(time.Hour) |
| + h.install("v3", "a") |
| + h.pass() |
| + time.Sleep(100 * time.Millisecond) |
| + if _, s, _ := h.act.snapshot(); len(s) != 1 { |
| + t.Fatalf("swaps started %v: an attempt was admitted after Close", s) |
| + } |
| +} |
| + |
| +// Task 10 review minor (1), restored in Task 11: the pointer pass requeues a |
| +// cleared build at once, before it reads the app's next channel, as the |
| +// armed pass did before the pointer work was shared (onCleared). Both |
| +// channels re-sign v2, so the second read always follows a clear. |
| +func TestAPointerClearRequeuesBeforeTheNextChannelIsRead(t *testing.T) { |
| + h := newHarness(t, "a", "b") |
| + h.stage("v2") |
| + rb := h.rows.get("b") |
| + rb.Channel = "beta" |
| + h.rows.put(rb) |
| + h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "c", PromotedAt: promotedAt, At: t0}) |
| + h.install("v2", "a", "b") |
| + h.pass() |
| + h.pointer("stable", "v2", promotedAt.Add(time.Minute)) |
| + h.pointer("beta", "v2", promotedAt.Add(time.Minute)) |
| + var reads atomic.Int32 |
| + var requeuedLate atomic.Bool |
| + h.mu.Lock() |
| + h.pointerHook = func() { |
| + if reads.Add(1) != 2 { |
| + return |
| + } |
| + h.m.mu.Lock() |
| + n := len(h.m.retried) |
| + h.m.mu.Unlock() |
| + if n == 0 { |
| + requeuedLate.Store(true) |
| + } |
| + } |
| + h.mu.Unlock() |
| + h.advance(pointerReadEvery) |
| + h.pass() |
| + if reads.Load() != 2 { |
| + t.Fatalf("%d pointer reads, want 2 (stable and beta)", reads.Load()) |
| + } |
| + if requeuedLate.Load() { |
| + t.Fatal("the cleared build was requeued only after every channel was read") |
| + } |
| + waitFor(t, "every host on v2", func() bool { return h.servingAll("v2", "a", "b") }) |
| +} |
| + |
| +// Task 11 Corrections (plan review #10): acks are ordered at the transition. |
| +// The writer keeps one pending ack per host, the newest by sequence; an older |
| +// one is dropped, never written after a newer one; hosts are written in the |
| +// order they were first queued; each write is bounded by ackWriteBound. |
| +func TestAcksAreWrittenNewestLastPerHost(t *testing.T) { |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + defer cancel() |
| + w := newAckWriter(ctx, func(string, ...any) {}) |
| + var mu sync.Mutex |
| + var got []string |
| + write := func(tag string) func(context.Context) { |
| + return func(ctx context.Context) { |
| + if dl, ok := ctx.Deadline(); !ok || time.Until(dl) > ackWriteBound { |
| + t.Errorf("ack %s written with deadline %v (%v), want at most ackWriteBound", tag, dl, ok) |
| + } |
| + mu.Lock() |
| + got = append(got, tag) |
| + mu.Unlock() |
| + } |
| + } |
| + entered, block := make(chan struct{}), make(chan struct{}) |
| + first := write("a1") |
| + w.submit("a", ackJob{seq: 1, write: func(ctx context.Context) { close(entered); <-block; first(ctx) }}) |
| + <-entered |
| + w.submit("a", ackJob{seq: 3, write: write("a3")}) |
| + w.submit("a", ackJob{seq: 2, write: write("a2")}) // older than the pending a3 |
| + w.submit("b", ackJob{seq: 4, write: write("b4")}) |
| + w.submit("a", ackJob{seq: 5, write: write("a5")}) // replaces a3, keeps a's place |
| + close(block) |
| + waitFor(t, "three writes", func() bool { mu.Lock(); defer mu.Unlock(); return len(got) == 3 }) |
| + w.submit("a", ackJob{seq: 2, write: write("a2-late")}) // older than one written |
| + time.Sleep(50 * time.Millisecond) |
| + mu.Lock() |
| + defer mu.Unlock() |
| + if want := []string{"a1", "a5", "b4"}; strings.Join(got, ",") != strings.Join(want, ",") { |
| + t.Fatalf("written %v, want %v", got, want) |
| + } |
| +} |
| + |
| +// Task 11 Corrections (spec §8.6): Close also joins a wake event under way. |
| +// A terminal OnGen writes the wake's outcome and ends its intent; Close does |
| +// not return while it is still running. |
| +func TestCloseWaitsForAnOnGenUnderWay(t *testing.T) { |
| + h := newHarness(t, "a") |
| + h.pass() |
| + entered, release := make(chan struct{}), make(chan struct{}) |
| + var once sync.Once |
| + h.rows.mu.Lock() |
| + h.rows.setServingHook = func(string) { |
| + once.Do(func() { close(entered); <-release }) |
| + } |
| + h.rows.mu.Unlock() |
| + g, _ := h.act.Current("a") |
| + go h.m.OnGen(activator.GenEvent{Kind: activator.GenSettled, Route: h.rows.get("a"), Gen: g, FirstWake: true}) |
| + <-entered |
| + closed := make(chan struct{}) |
| + go func() { |
| + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) |
| + defer cancel() |
| + h.m.Close(ctx) |
| + close(closed) |
| + }() |
| + select { |
| + case <-closed: |
| + t.Fatal("Close returned while a wake event was still being recorded") |
| + case <-time.After(100 * time.Millisecond): |
| + } |
| + close(release) |
| + select { |
| + case <-closed: |
| + case <-time.After(5 * time.Second): |
| + t.Fatal("Close did not return once the wake event finished") |
| + } |
| +} |
| + |
| +// Task 11 Corrections: closing the writer writes what is pending, each |
| +// write still bounded, and an ack submitted after it is dropped. |
| +func TestClosingTheAckWriterWritesWhatIsPending(t *testing.T) { |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + defer cancel() |
| + w := newAckWriter(ctx, func(string, ...any) {}) |
| + var mu sync.Mutex |
| + var got []string |
| + write := func(tag string) func(context.Context) { |
| + return func(context.Context) { mu.Lock(); got = append(got, tag); mu.Unlock() } |
| + } |
| + entered, block := make(chan struct{}), make(chan struct{}) |
| + w.submit("a", ackJob{seq: 1, write: func(context.Context) { close(entered); <-block }}) |
| + <-entered |
| + w.submit("b", ackJob{seq: 2, write: write("b2")}) |
| + w.submit("c", ackJob{seq: 3, write: write("c3")}) |
| + go func() { time.Sleep(50 * time.Millisecond); close(block) }() |
| + cctx, ccancel := context.WithTimeout(context.Background(), 5*time.Second) |
| + defer ccancel() |
| + w.close(cctx) |
| + w.submit("d", ackJob{seq: 4, write: write("d4")}) |
| + time.Sleep(50 * time.Millisecond) |
| + mu.Lock() |
| + defer mu.Unlock() |
| + if strings.Join(got, ",") != "b2,c3" { |
| + t.Fatalf("written %v, want b2 and c3 written before close returned, and nothing after", got) |
| + } |
| +} |
diff --git a/internal/rollout/policy.go b/internal/rollout/policy.go| index 2c780b224..2d86cb2b7 100644 |
| --- a/internal/rollout/policy.go |
| +++ b/internal/rollout/policy.go |
| @@ -62,43 +62,90 @@ func RecoveryMode(p AppPolicy, retained string) activator.SwapMode { |
| return AttemptMode(p, retained) |
| } |
| +// policyEntry is one app's cached policy. fills numbers the reads begun for |
| +// the app and stored is the number of the read whose result is held, so an |
| +// older read that finishes after a newer one has stored leaves the newer |
| +// value in place (Astra R1 Flag triage T3). failedAt is the last failed |
| +// read's time, zero once a read has succeeded since. |
| type policyEntry struct { |
| - p AppPolicy |
| - at time.Time |
| + p AppPolicy |
| + ok bool // p was read, at at |
| + at time.Time |
| + failedAt time.Time |
| + fills uint64 |
| + stored uint64 |
| } |
| // policyCache reads each app's policy at most once per ttl. A failed read |
| // keeps the last value it had; an app whose policy this box has never read |
| // gets hold, because a guess of overlap could run a migration beside a build |
| -// that cannot share the database. |
| +// that cannot share the database. A failed read is not tried again for |
| +// policyRetryAfter, so a store that is failing costs one read per app per |
| +// backoff rather than one per pass, wake and attempt (Task 11 rule 1). |
| type policyCache struct { |
| load func(ctx context.Context, acct, app string) (*appconfig.Manifest, error) |
| ttl time.Duration |
| now func() time.Time |
| mu sync.Mutex |
| - m map[appKey]policyEntry |
| + m map[appKey]*policyEntry |
| } |
| +// policyRetryAfter is how long a failed policy read waits before the next. |
| +const policyRetryAfter = 5 * time.Second |
| + |
| +// readBound bounds each store read the scheduler makes in line (the policy, |
| +// the channel pointer, the canonical demand): a store that hangs costs a |
| +// pass, a wake plan or an attempt this long, not forever (Task 11 rule 2, |
| +// Astra R1 Flag triage T8 and T10). |
| +const readBound = 5 * time.Second |
| + |
| +// get is the app's policy. A missing bundle is not an error here: |
| +// appconfig.Get answers it with an empty manifest, which is the defaults. An |
| +// error is never the defaults. |
| func (c *policyCache) get(ctx context.Context, acct, app string) AppPolicy { |
| k := appKey{acct, app} |
| c.mu.Lock() |
| - e, seen := c.m[k] |
| - c.mu.Unlock() |
| - if seen && c.now().Sub(e.at) < c.ttl { |
| - return e.p |
| + e := c.m[k] |
| + if e == nil { |
| + e = &policyEntry{} |
| + c.m[k] = e |
| } |
| - m, err := c.load(ctx, acct, app) |
| - if err != nil { |
| - if seen { |
| - return e.p |
| - } |
| - return AppPolicy{Swap: activator.ModeHold, Concurrency: 1} |
| + now := c.now() |
| + if e.ok && now.Sub(e.at) < c.ttl { |
| + p := e.p |
| + c.mu.Unlock() |
| + return p |
| } |
| - p := PolicyFromManifest(m) |
| - c.mu.Lock() |
| - c.m[k] = policyEntry{p: p, at: c.now()} |
| + if !e.failedAt.IsZero() && now.Sub(e.failedAt) < policyRetryAfter { |
| + p := e.lastOrHold() |
| + c.mu.Unlock() |
| + return p |
| + } |
| + e.fills++ |
| + fill := e.fills |
| c.mu.Unlock() |
| - return p |
| + lctx, cancel := context.WithTimeout(ctx, readBound) |
| + m, err := c.load(lctx, acct, app) |
| + cancel() |
| + c.mu.Lock() |
| + defer c.mu.Unlock() |
| + if fill > e.stored { |
| + // No newer read has stored since this one began. |
| + if err != nil { |
| + e.failedAt = c.now() |
| + } else { |
| + e.p, e.ok, e.at, e.failedAt, e.stored = PolicyFromManifest(m), true, c.now(), time.Time{}, fill |
| + } |
| + } |
| + return e.lastOrHold() |
| +} |
| + |
| +// lastOrHold is the last policy read, or hold for an app never read. |
| +func (e *policyEntry) lastOrHold() AppPolicy { |
| + if e.ok { |
| + return e.p |
| + } |
| + return AppPolicy{Swap: activator.ModeHold, Concurrency: 1} |
| } |
| // Managed reports whether rt is an activator-managed exec route this |
diff --git a/internal/rollout/policy_test.go b/internal/rollout/policy_test.go| index 43d40e10d..570fbf31b 100644 |
| --- a/internal/rollout/policy_test.go |
| +++ b/internal/rollout/policy_test.go |
| @@ -3,6 +3,7 @@ package rollout |
| import ( |
| "context" |
| "errors" |
| + "sync" |
| "testing" |
| "time" |
| @@ -69,13 +70,14 @@ func TestPolicyCacheNeverGuessesOverlap(t *testing.T) { |
| }, |
| ttl: 30 * time.Second, |
| now: func() time.Time { return now }, |
| - m: map[appKey]policyEntry{}, |
| + m: map[appKey]*policyEntry{}, |
| } |
| ctx := context.Background() |
| fail = errors.New("SlowDown") |
| if got := c.get(ctx, "bab", "blog"); got.Swap != activator.ModeHold { |
| t.Fatalf("never-read policy = %v, want hold", got.Swap) |
| } |
| + now = now.Add(policyRetryAfter) // the failed read's backoff (Task 11) |
| fail, answer = nil, &appconfig.Manifest{SwapConcurrency: 2} |
| if got := c.get(ctx, "bab", "blog"); got != (AppPolicy{Swap: activator.ModeOverlap, Concurrency: 2}) { |
| t.Fatalf("first read = %+v", got) |
| @@ -91,6 +93,101 @@ func TestPolicyCacheNeverGuessesOverlap(t *testing.T) { |
| } |
| } |
| +// Task 11 rule 1 (Astra R1 Flag triage T3): two fills of one app's policy |
| +// can overlap, and the older one can finish last. Its result is older than |
| +// what the newer fill stored, so it must not replace it. |
| +func TestPolicyCacheAnOlderFillNeverOverwritesANewerOne(t *testing.T) { |
| + now := time.Unix(1_700_000_000, 0) |
| + entered, release := make(chan struct{}), make(chan struct{}) |
| + var mu sync.Mutex |
| + calls := 0 |
| + c := &policyCache{ |
| + load: func(ctx context.Context, _, _ string) (*appconfig.Manifest, error) { |
| + if _, ok := ctx.Deadline(); !ok { |
| + t.Error("a policy load ran with no deadline (readBound)") |
| + } |
| + mu.Lock() |
| + calls++ |
| + n := calls |
| + mu.Unlock() |
| + if n == 1 { |
| + close(entered) |
| + <-release |
| + return &appconfig.Manifest{Swap: "hold"}, nil |
| + } |
| + return &appconfig.Manifest{SwapConcurrency: 3}, nil |
| + }, |
| + ttl: 30 * time.Second, |
| + now: func() time.Time { return now }, |
| + m: map[appKey]*policyEntry{}, |
| + } |
| + ctx := context.Background() |
| + older := make(chan AppPolicy) |
| + go func() { older <- c.get(ctx, "bab", "blog") }() |
| + <-entered |
| + newer := AppPolicy{Swap: activator.ModeOverlap, Concurrency: 3} |
| + if got := c.get(ctx, "bab", "blog"); got != newer { |
| + t.Fatalf("the newer fill = %+v, want %+v", got, newer) |
| + } |
| + close(release) |
| + if got := <-older; got != newer { |
| + t.Fatalf("the older fill, finishing last, answered %+v; want the newer stored %+v", got, newer) |
| + } |
| + if got := c.get(ctx, "bab", "blog"); got != newer { |
| + t.Fatalf("after the older fill finished = %+v, want the newer fill's %+v", got, newer) |
| + } |
| + if calls != 2 { |
| + t.Fatalf("%d loads, want 2: the stored value is fresh", calls) |
| + } |
| +} |
| + |
| +// Task 11 rule 1 (Astra R1 Flag triage T3): a failed read is not retried for |
| +// policyRetryAfter, whether or not a value was ever read. Meanwhile the |
| +// cache answers its last value, or hold for an app never read. A missing |
| +// bundle (appconfig.Get's empty manifest) is the defaults; an error never is. |
| +func TestPolicyCacheBacksOffAFailedRead(t *testing.T) { |
| + now := time.Unix(1_700_000_000, 0) |
| + calls := 0 |
| + var answer *appconfig.Manifest |
| + fail := errors.New("SlowDown") |
| + c := &policyCache{ |
| + load: func(context.Context, string, string) (*appconfig.Manifest, error) { |
| + calls++ |
| + return answer, fail |
| + }, |
| + ttl: 30 * time.Second, |
| + now: func() time.Time { return now }, |
| + m: map[appKey]*policyEntry{}, |
| + } |
| + ctx := context.Background() |
| + hold := AppPolicy{Swap: activator.ModeHold, Concurrency: 1} |
| + for i := range 3 { |
| + if got := c.get(ctx, "bab", "blog"); got != hold { |
| + t.Fatalf("never-read policy, read %d = %+v, want hold", i, got) |
| + } |
| + now = now.Add(time.Second) |
| + } |
| + if calls != 1 { |
| + t.Fatalf("%d loads inside policyRetryAfter of a failure, want 1", calls) |
| + } |
| + now = now.Add(policyRetryAfter) |
| + fail, answer = nil, &appconfig.Manifest{Schema: 1, App: "blog"} // missing: appconfig.Get's empty manifest |
| + if got := c.get(ctx, "bab", "blog"); got != DefaultPolicy() || calls != 2 { |
| + t.Fatalf("after the backoff = %+v (%d loads), want the defaults from a second load", got, calls) |
| + } |
| + now = now.Add(31 * time.Second) |
| + fail = errors.New("SlowDown") |
| + for i := range 3 { |
| + if got := c.get(ctx, "bab", "blog"); got != DefaultPolicy() { |
| + t.Fatalf("stale policy with a failing read, read %d = %+v, want the last known", i, got) |
| + } |
| + now = now.Add(time.Second) |
| + } |
| + if calls != 3 { |
| + t.Fatalf("%d loads, want 3: one failed refresh, then none inside policyRetryAfter", calls) |
| + } |
| +} |
| + |
| func TestManagedIsTheActivatorsExecRoutes(t *testing.T) { |
| cases := []struct { |
| rt registry.Route |
diff --git a/internal/rollout/scheduler.go b/internal/rollout/scheduler.go| index 4d102226f..f4c8d7904 100644 |
| --- a/internal/rollout/scheduler.go |
| +++ b/internal/rollout/scheduler.go |
| @@ -46,6 +46,10 @@ func (s *scheduler) loop(ctx context.Context) { |
| case <-s.kick: |
| case <-timer.C: |
| } |
| + if s.m.closing() { |
| + // The agent is stopping (Manager.Close): nothing more is admitted. |
| + return |
| + } |
| view, pendingConfig := s.view() |
| for host, rev := range pendingConfig { |
| if serving, seen := view[host]; seen && serving == "" { |
| @@ -58,7 +62,17 @@ func (s *scheduler) loop(ctx context.Context) { |
| next := s.q.nextDue(now) |
| s.mu.Unlock() |
| for _, d := range ds { |
| - go s.execute(ctx, d) |
| + // Counted before it starts, so Close waits for its done(). One |
| + // admitted as Close began is not run: nothing of it has been |
| + // written yet, and its slot stays taken, which keeps every later |
| + // admission for the host out until the process exits. |
| + if !s.m.enter(true) { |
| + continue |
| + } |
| + go func(d dispatch) { |
| + defer s.m.leave() |
| + s.execute(ctx, d) |
| + }(d) |
| } |
| wait := time.Hour |
| if !next.IsZero() { |
| @@ -153,7 +167,7 @@ func (s *scheduler) execute(ctx context.Context, d dispatch) { |
| // The failing pointer's PromotedAt, read now, while nothing is at stake: |
| // done() publishes a failure without a blocking read (spec §8.1). |
| if d.Class == classAdopt { |
| - d.FailedUnder = m.failedUnder(*rt, d.ID.Build) |
| + d.FailedUnder = m.failedUnder(ctx, *rt, d.ID.Build) |
| } |
| // The execution's start, after every read and before the activator runs |
| // anything: a hold-ran marker this attempt leaves carries it (ruling 205), |
diff --git a/internal/rollout/wake_token_test.go b/internal/rollout/wake_token_test.go| index 4c241c864..84384d9aa 100644 |
| --- a/internal/rollout/wake_token_test.go |
| +++ b/internal/rollout/wake_token_test.go |
| @@ -93,7 +93,7 @@ func TestAStaleOutcomeLeavesANewerWakesObservationAndConfig(t *testing.T) { |
| h.registerThenInstall("v2", "a") |
| h.wakeHost("a") |
| var acks []appconfig.ConfigAck |
| - ack := h.m.DeferAck(func(a appconfig.ConfigAck) { acks = append(acks, a) }) |
| + ack := h.m.DeferAck(func(_ context.Context, a appconfig.ConfigAck) { acks = append(acks, a) }) |
| h.m.ConfigDelivered("a", "sha256:c1") |
| old := h.m.Plan(ctx, h.rows.get("a")) |
| @@ -106,6 +106,7 @@ func TestAStaleOutcomeLeavesANewerWakesObservationAndConfig(t *testing.T) { |
| g := activator.Gen{N: 7, Build: "v2", Digest: "sha-v2"} |
| h.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: h.rows.get("a"), Gen: g, FirstWake: true, |
| Token: old.Token, ConfigRev: old.ConfigRev}) |
| + h.flushAcks() |
| if len(acks) != 0 { |
| t.Fatalf("acks %+v: an older wake's late event acknowledged a revision", acks) |
| } |
| @@ -115,6 +116,7 @@ func TestAStaleOutcomeLeavesANewerWakesObservationAndConfig(t *testing.T) { |
| h.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: h.rows.get("a"), Gen: g, FirstWake: true, |
| Token: cur.Token, ConfigRev: cur.ConfigRev}) |
| + h.flushAcks() |
| if len(acks) != 1 { |
| t.Fatalf("acks %+v after the current wake's healthy event, want c2's", acks) |
| } |
diff --git a/internal/rollout/wakeplan.go b/internal/rollout/wakeplan.go| index 3500693e3..06675ba58 100644 |
| --- a/internal/rollout/wakeplan.go |
| +++ b/internal/rollout/wakeplan.go |
| @@ -345,7 +345,7 @@ func (m *Manager) planRow(ctx context.Context, rt registry.Route, tok uint64) ac |
| if plan.FirstWake && plan.Boot.Version != "" { |
| // Read now, before the wake runs, so a failure of the build is |
| // published with no blocking read (buildFailed, spec §8.1). |
| - plan.FailedUnder = m.failedUnder(rt, plan.Boot.Version) |
| + plan.FailedUnder = m.failedUnder(ctx, rt, plan.Boot.Version) |
| } |
| if plan.Boot.StagedPath != "" { |
| plan.Check = m.planCheck(rt, plan) |
| @@ -441,6 +441,10 @@ func (m *Manager) attemptRunning(host string) bool { |
| // good build is published only while its generation is the one the |
| // activator serves (acknowledged). A token of zero matches nothing. |
| func (m *Manager) OnGen(ev activator.GenEvent) { |
| + // Joined by Close (spec §8.6): a wake's terminal event writes its outcome |
| + // and ends its intent, which must be on disk before the agent exits. |
| + m.enter(false) |
| + defer m.leave() |
| rt := ev.Route |
| // The wake's start, its plan: a hold-ran marker it leaves carries the |
| // plan time (ruling 205), and its good write and its failure are ordered |
| @@ -552,8 +556,9 @@ func (m *Manager) wakeConfigActivated(host string, tok uint64, rev string) { |
| return |
| } |
| write, _ := m.track(host).activate(rev) |
| + ack := m.ackLocked(write) |
| m.mu.Unlock() |
| - m.activationDone(host, rev, write) |
| + m.activationDone(host, rev, ack) |
| } |
| // buildFailed is a failed first wake's cohort-failure transition (spec |
diff --git a/internal/rollout/wakeplan_test.go b/internal/rollout/wakeplan_test.go| index eb65428b9..26bec7f50 100644 |
| --- a/internal/rollout/wakeplan_test.go |
| +++ b/internal/rollout/wakeplan_test.go |
| @@ -263,9 +263,10 @@ func TestAWakeActivatesTheConfigItStartedWith(t *testing.T) { |
| h.act.owned["a"] = true // waking: a process, not yet current |
| h.act.mu.Unlock() |
| var acks int |
| - ack := h.m.DeferAck(func(appconfig.ConfigAck) { acks++ }) |
| + ack := h.m.DeferAck(func(context.Context, appconfig.ConfigAck) { acks++ }) |
| h.m.ConfigDelivered("a", "sha256:c1") |
| ack(appconfig.ConfigAck{Host: "a"}) |
| + h.flushAcks() |
| if acks != 0 { |
| t.Fatal("acknowledged before any generation started with the config") |
| } |
| @@ -274,6 +275,7 @@ func TestAWakeActivatesTheConfigItStartedWith(t *testing.T) { |
| // The activator carries the plan's token and revision onto the event. |
| h.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: rt, Gen: activator.Gen{Build: "v1", Digest: "sha-v1"}, |
| Token: p.Token, ConfigRev: p.ConfigRev}) |
| + h.flushAcks() |
| if acks != 1 { |
| t.Fatalf("acks = %d after the wake, want 1", acks) |
| } |
| @@ -289,9 +291,10 @@ func TestASleepersConfigNeedsNoAttempt(t *testing.T) { |
| delete(h.act.gens, "a") // asleep: no process at all |
| h.act.mu.Unlock() |
| var acks []appconfig.ConfigAck |
| - ack := h.m.DeferAck(func(a appconfig.ConfigAck) { acks = append(acks, a) }) |
| + ack := h.m.DeferAck(func(_ context.Context, a appconfig.ConfigAck) { acks = append(acks, a) }) |
| h.m.ConfigDelivered("a", "sha256:c1") |
| ack(appconfig.ConfigAck{Host: "a", Restarted: true}) |
| + h.flushAcks() |
| if len(acks) != 1 || !acks[0].Hibernating || acks[0].Restarted { |
| t.Fatalf("acks = %+v, want one at delivery, Hibernating", acks) |
| } |