a73x

2121dde5

fix(agent): narration cannot outlive the pass that said it

a73x   2026-08-12 14:39

Commit message
fix(agent): narration cannot outlive the pass that said it

A create narrates itself from goroutines it does not own: the image cache
shares one download between waiters and runs the progress callback on the
fetch. When a create is abandoned at VMTimeout, that fetch keeps draining,
and one late progress line landed on the worker's result AFTER the pass had
published its failure — putting "creating — downloading image" back on top
of a settled failed row. The next host report showed a phantom in-flight
create and hid the failure.

Each pass now carries a generation, stamped by the worker loop and closed
over by the publisher it hands down. A mid-pass publish lands only while
that pass is still the one in flight, so a straggler after the pass-end
publish and a straggler from an earlier pass are both dropped, under the
same lock the pass-end publish takes. A create also stops narrating once
its own context is done, so a finished pass is not called into at all.

Narration is commentary, not state — now by construction rather than by
timing.
internal/agent/reconcile/narrate_test.go
Old New
@@ -2,6 +2,8 @@ package reconcile
2 2
3 import ( 3 import (
4 "context" 4 "context"
5 "errors"
6 "sync"
5 "testing" 7 "testing"
6 8
7 "github.com/a73x/eitri/internal/agent/state" 9 "github.com/a73x/eitri/internal/agent/state"
@@ -79,6 +81,81 @@ func TestACreateWithNobodyListeningStillCreates(t *testing.T) {
79 assert.Empty(t, res.vm.GetStatusDetail(), "a settled VM has nothing to add") 81 assert.Empty(t, res.vm.GetStatusDetail(), "a settled VM has nothing to add")
80 } 82 }
81 83
84 // TestNarrationCannotOutliveThePassThatSaidIt is the straggler case. The image
85 // cache shares one fetch between waiters, so its progress callback runs on a
86 // goroutine the pass does not own and keeps draining after the pass has given
87 // up on the download. A line landing after the pass published its failure would
88 // replace a settled "failed" row with "creating — downloading image", and the
89 // next host report would show a phantom in-flight create where a failure
90 // happened — narration would have become state.
91 //
92 // It also pins the feature the guard could quietly kill: while the pass is
93 // still live, its narration must land in the report as it always did.
94 func TestNarrationCannotOutliveThePassThatSaidIt(t *testing.T) {
95 f := setup(t)
96 f.eng.MaxCreateAttempts = 1 // one failure is terminal, so the final row is unambiguous
97
98 var mu sync.Mutex
99 var straggler func(done, total int64)
100 narrated := make(chan struct{})
101 release := make(chan struct{})
102 f.eng.Images = func(_ context.Context, _, _ string, progress func(done, total int64)) (string, error) {
103 mu.Lock()
104 straggler = progress // kept past the fetch, the way a shared fetch keeps it
105 mu.Unlock()
106 progress(1_288_490_188, 3_972_844_748)
107 close(narrated)
108 <-release
109 return "", errors.New("fetch gave up")
110 }
111
112 stepNoWait(t, f, snap(1, vm("vm1")))
113 <-narrated
114
115 row := findVM(f.aggregateNow(1), "vm1")
116 require.NotNil(t, row, "a live pass still narrates")
117 assert.Equal(t, "downloading image 1.2/3.7 GiB", row.GetStatusDetail())
118
119 close(release)
120 f.eng.manager().waitIdle()
121 require.Equal(t, "failed", findVM(f.aggregateNow(1), "vm1").GetPhase(), "precondition: the pass ended failed")
122
123 // The fetch goroutine drains on and says one more thing, after the pass that
124 // asked for it has published its failure.
125 mu.Lock()
126 say := straggler
127 mu.Unlock()
128 say(3_972_844_748, 3_972_844_748)
129
130 final := findVM(f.aggregateNow(1), "vm1")
131 require.NotNil(t, final)
132 assert.Equal(t, "failed", final.GetPhase(), "a straggler must not resurrect a finished create")
133 assert.Equal(t, "fetch gave up", final.GetLastError())
134 assert.Empty(t, final.GetStatusDetail())
135 }
136
137 // TestADeadPassSaysNothing: the source-side half. A pass whose context has
138 // expired stops narrating at all, so the goroutine still draining a shared
139 // fetch never calls into a pass that has given up on it.
140 func TestADeadPassSaysNothing(t *testing.T) {
141 f := setup(t)
142 ctx, cancel := context.WithCancel(context.Background())
143 defer cancel()
144
145 f.eng.Images = func(_ context.Context, _, sha string, progress func(done, total int64)) (string, error) {
146 cancel() // the watchdog fires while the fetch is still running
147 progress(1_288_490_188, 3_972_844_748)
148 return "/cache/" + sha + ".raw", nil
149 }
150
151 var said []string
152 var res vmResult
153 f.eng.create(ctx, vm("vm1"), state.Record{}, false, &res,
154 func(interim vmResult) { said = append(said, interim.vm.GetStatusDetail()) })
155
156 assert.NotContains(t, said, "downloading image 1.2/3.7 GiB")
157 }
158
82 func TestDownloadDetailWords(t *testing.T) { 159 func TestDownloadDetailWords(t *testing.T) {
83 assert.Equal(t, "downloading image 1.2/3.7 GiB", downloadDetail(1_288_490_188, 3_972_844_748)) 160 assert.Equal(t, "downloading image 1.2/3.7 GiB", downloadDetail(1_288_490_188, 3_972_844_748))
84 // Both halves scale together off the larger, so the pair reads as a fraction 161 // Both halves scale together off the larger, so the pair reads as a fraction
internal/agent/reconcile/reconcile.go
Old New
@@ -739,8 +739,17 @@ func (e *Engine) create(ctx context.Context, d *pb.VMDesired, rec state.Record,
739 // level-triggered — a row that dropped the key would ask the control plane 739 // level-triggered — a row that dropped the key would ask the control plane
740 // to forget a guest's certificate for a tick — and rec is read at each call, 740 // to forget a guest's certificate for a tick — and rec is read at each call,
741 // so a step after Boot narrates with the address Boot found. 741 // so a step after Boot narrates with the address Boot found.
742 //
743 // A dead pass says nothing. say runs on whatever goroutine calls it, and the
744 // image cache's progress callback runs on a shared fetch goroutine that
745 // keeps draining after this pass abandons the download at VMTimeout — this
746 // stops that goroutine calling into a finished pass at all. It is the polite
747 // half of the guarantee, not the load-bearing one: ctx expiry and the pass
748 // ending are not the same instant, and a pass with no VMTimeout has a
749 // context that never expires. What actually makes a stale row unlandable is
750 // the publisher's generation check (see worker.publish).
742 say := func(detail string) { 751 say := func(detail string) {
743 if pub == nil { 752 if pub == nil || ctx.Err() != nil {
744 return 753 return
745 } 754 }
746 var interim vmResult 755 var interim vmResult
internal/agent/reconcile/worker.go
Old New
@@ -165,6 +165,10 @@ type worker struct {
165 pending *assignment 165 pending *assignment
166 // busy reports that a pass is running (with mu released). 166 // busy reports that a pass is running (with mu released).
167 busy bool 167 busy bool
168 // gen names the pass currently in flight. run stamps a fresh one before
169 // every pass and hands it to that pass's publisher, so a row can be traced
170 // back to the pass that produced it — see publish.
171 gen uint64
168 // stopped ends the goroutine after the current pass. 172 // stopped ends the goroutine after the current pass.
169 stopped bool 173 stopped bool
170 // result is this VM's last-published contribution to the host report. 174 // result is this VM's last-published contribution to the host report.
@@ -196,11 +200,16 @@ func (w *worker) run() {
196 a := *w.pending 200 a := *w.pending
197 w.pending = nil 201 w.pending = nil
198 w.busy = true 202 w.busy = true
203 w.gen++
204 gen := w.gen
199 w.mu.Unlock() 205 w.mu.Unlock()
200 206
201 // LOAD-BEARING: the pass runs with w.mu RELEASED. That is what lets 207 // LOAD-BEARING: the pass runs with w.mu RELEASED. That is what lets
202 // deliver and collect run at full speed while this VM does slow work. 208 // deliver and collect run at full speed while this VM does slow work.
203 res, ok := w.eng.reconcileOne(context.Background(), w.id, a, w.publish) 209 // The publisher carries this pass's generation, which is what stops a
210 // row from a finished pass landing on a later one (see publish).
211 res, ok := w.eng.reconcileOne(context.Background(), w.id, a,
212 func(interim vmResult) { w.publish(gen, interim) })
204 213
205 w.mu.Lock() 214 w.mu.Lock()
206 // Publish only a pass that ran. A skipped pass (this VM's record could 215 // Publish only a pass that ran. A skipped pass (this VM's record could
@@ -212,6 +221,9 @@ func (w *worker) run() {
212 if ok { 221 if ok {
213 w.result = res 222 w.result = res
214 } 223 }
224 // Clearing busy under the same lock that lands the result is what
225 // closes the pass to further publishes: a straggler that takes the
226 // lock after this point finds no pass in flight and is dropped.
215 w.busy = false 227 w.busy = false
216 w.cond.Broadcast() 228 w.cond.Broadcast()
217 w.mu.Unlock() 229 w.mu.Unlock()
@@ -225,12 +237,28 @@ func (w *worker) run() {
225 // minutes — a first create. 237 // minutes — a first create.
226 // 238 //
227 // It takes the same lock the pass-end publish takes and holds it no longer, so 239 // It takes the same lock the pass-end publish takes and holds it no longer, so
228 // a narrating create is no more able to delay a report than a silent one. The 240 // a narrating create is no more able to delay a report than a silent one.
229 // pass's own result lands on top when it ends, which is what keeps this 241 //
230 // commentary and not state: nothing here can outlive the pass that said it. 242 // GENERATION RULE: gen is the generation run stamped on the pass that was
231 func (w *worker) publish(res vmResult) { 243 // handed this publisher, and a row lands only if that pass is still the one in
244 // flight. Narration reaches here on goroutines the pass does not own — the
245 // image cache shares one download between waiters and runs its progress
246 // callback on the fetch — so a callback can outlive the pass that asked for
247 // one: a create abandoned at VMTimeout returns and publishes "failed" while the
248 // fetch keeps draining, and one late progress line would otherwise put
249 // "creating — downloading image" back on top of that settled row. The host
250 // report would then show a phantom in-flight create and hide the failure. Both
251 // stale orders die here: a callback after the pass-end publish finds busy
252 // false, and a callback from an earlier pass while a later one runs finds a
253 // newer gen. A callback that beats the pass-end publish is admitted and simply
254 // overwritten by it — the ordering that already made this commentary and not
255 // state: nothing here can outlive the pass that said it.
256 func (w *worker) publish(gen uint64, res vmResult) {
232 w.mu.Lock() 257 w.mu.Lock()
233 defer w.mu.Unlock() 258 defer w.mu.Unlock()
259 if !w.busy || gen != w.gen {
260 return // a straggler from a pass that has already ended
261 }
234 w.result = res 262 w.result = res
235 } 263 }
236 264