Skip to content

Commit 45332fa

Browse files
tests: a reading beside the work is waited on, not guessed at (#952)
* tests: a reading beside the work is waited on, not guessed at Five tests were green alone and red beside another suite, a different one each run, none of them on the known-red ledger. Four causes, one shape: a test measured the scheduler and called it the road. A HELD MEMORY LINE IS NO LONGER LOST, and this one was the product's. `refreshMemory` drains the lines the post-turn pass had no stream for, and it said them onto the hub the reading was handed when it started. A recall rides beside the turn and is never joined, so on a busy machine it runs after its turn has ended: the line was taken off the queue and said to a closed stream, and never said again. Instrumented on this box, every losing run logged `ctxErr=context canceled hubClosed=true`. Every memory line now goes through `Agent.sayMemory` — the one door that says it on whichever stream is live and holds it when there is none — and that send happens under `a.mu`, the lock every road that ends a turn clears `a.hub` under before it closes the hub, and that a starting turn holds until it has subscribed. So there is no instant at which a line is sent onto a stream that has closed. `memoryNotice` and the captured hubs are gone with it: `routedMemory` and `runMemoryCommand` take `say func(string)`, where nil is a node's reading borrowing its parent's store and saying nothing, as it always did. THREE FIXTURES WAITED ON A FIGURE AND NOW WAIT ON THE READING. A scripted conversation answers in no time and never blocks, so a reading the turn has just started can sit unscheduled behind it for as many rounds as the machine is busy — the mark's drawing was still pending at all eight remaining boundaries in every failing run, the script then ended in words, and `takeAtTheEnd` journalled a drawing nobody could spend. `readBeside` now tells a watch carried on the work's context when each reading under it starts and when it has landed, after the settle, and a fixture that opts in through `watchReadings` answers the conversation only once none is in flight. That is the order a real turn usually has: a real model spends seconds on a step, the readings beside it are small, and a drawing that splits cuts the step it lands in. `routeCompleter.holdUntilRaced` and its millisecond spin (`awaitRace`, `stopHolding`) are deleted — a call entered is a verdict still to be parsed and settled, which is what the boundary after the first word kept missing. THE PRODUCT CARRIES NO WATCH AND NEVER WAITS ON ONE, and `TestTheRunningProductNeverWaitsOnAReadingsLanding` (go/ast, so it is on the laws gate) fails the build on the two ways it would reach for one. A ROUTER'S FIRST TOKEN IS ORDERED BY A SIGNAL, NOT BY TEN MILLISECONDS. `TestARescueTheAllowanceRefusedSaysSoOnTheRow` scripted lane A at sixty milliseconds against a ceiling of fifty, so a starved timer let the first token arrive first and the row said nothing had been done at all. The stub takes a clock of the scenario's own (`laneRig.answersAfterTheWord`) whose waits end no sooner than the controller has spoken — `lanestub`'s own `StallUntil` rule, which cannot hold a first token — and a switch releases it too, so a regression that sends the refused rescue fails on the rescue rather than hanging. Proof, on the Spark with twelve CPU burners beside it and GOMAXPROCS=2: the four session reds and the new pin, `-count=50`, 250/250 green (about one failure in fifteen before); the same set `-count=20 -race`, 120/120 and no data race; all 211 tests of the changed files green, and again under -race; `TestARescueTheAllowanceRefusedSaysSoOnTheRow` `-count=500` green, where it failed once in three hundred before. docs/rules/ci.md says what to do with a test that fails only beside another suite, since that is where the known-red ledger's rules already live. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * Change entry for #952 Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * tests: a lane holds its first word on a signal, the way it already holds a stall The rule that a scripted arm orders itself by a signal and never by elapsed time had no way to reach a FIRST token: lanestub's StallUntil only takes hold after StallAfter deltas. So the rule lived in the provider's own test file as a clock the scenario installed through Server.SetClock, which delayed every wait the stub spent, for every lane, to order one lane's first token. It is a knob beside StallUntil now. Profile.FirstTokenUntil holds the lane's first word until a channel closes, after the scripted TTFT is spent, and holds the whole reply on the unstreamed path where the first token and the last arrive together. Its zero value is today's behaviour, so nothing that only names a duration changes. The refusal test hands lane A theControllersWord: one OnPhase reader that closes on any rung of the waiting ladder — either half of control.Report, a rescue, a model switch — so a regression that acts differently fails on its assertion rather than hanging on the signal. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * the hub says whether a line landed, so holding one needs no ordering A memory line that finds no live stream is held for the next turn, and until now "finds no live stream" meant "a.hub was nil while a.mu was held" — which is only true because every road that ends a turn clears that pointer under the lock before closing the hub. It is true today on all three roads and nothing enforces it. The day a fourth road closes before it clears, the defect this change fixes comes back in exactly the same shape, silently. So eventHub.send reports whether the event landed. Only the hub's own lock can answer that — a hub reads open until the moment it closes under it, so anything a caller checks beforehand is already a guess — and sayMemory now holds the line whenever the answer is no, including when there was no hub at all, which the same door answers. The ALL-CAPS paragraph that used to carry the correctness is now an explanation of why holding a line is rare. Three more from the same review. quiet takes the context the readings were started under, so a reading whose call ignores its cancellation can no longer park the next caller forever and report as a package timeout naming whichever test was running. theControllersWord takes the reader it chains to BEFORE installing the chain, where it used to assign it afterwards — a write racing its own reads by the memory model, in the one fixture whose job is to kill races. And startRecallLocked drops the hub parameter it stopped using. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
1 parent b4617e2 commit 45332fa

14 files changed

Lines changed: 586 additions & 150 deletions

File tree

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
---
2+
kind: fixed
3+
title: A reading beside the work is waited on, not guessed at, and a held memory line is not lost
4+
pr: 952
5+
surface: [chat, engine]
6+
invalidates:
7+
- "A HELD MEMORY LINE IS NO LONGER SAID INTO THE DARK. `Agent.refreshMemory` took the lines the post-turn pass had no stream for off the queue and said them onto the hub the reading had been HANDED when it started (`memoryNotice(hub, line)`). A recall rides beside the turn and is never joined, so on a busy machine it runs after that turn has ended — measured on this package's own suite beside another one, where every losing run logged `ctxErr=context canceled hubClosed=true` — and the line was taken, said to a closed stream, and never said again. Every memory line now goes through `Agent.sayMemory`, which asks the hub to say it and HOLDS IT WHEN THE HUB COULD NOT: `eventHub.send` reports whether the event landed, so a line that met a closed stream, or no stream at all, is queued for the next turn by the same door. The correctness of that no longer rests on any ordering between this door and the roads that end turns — the hub answers from under its own lock, which is the only place the answer is not already stale."
8+
- "`memoryNotice` IS GONE, AND SO ARE THE CAPTURED HUBS. `Agent.routedMemory` and `Agent.runMemoryCommand` take `say func(string)` where they took `hub *eventHub`, and nil says nothing — which is a task node's reading borrowing its parent's store, exactly as silent as it was. `Agent.refreshMemory(ctx, cue)`, `Agent.importMemoryFile()` and `Agent.startRecallLocked(ctx, cue)` no longer take a hub at all; nothing in the memory pass holds one."
9+
- "`eventHub.send` REPORTS WHETHER THE EVENT LANDED (`send(Event) (landed bool)`, agent.go). It answered nothing before, so a caller holding something a person is owed had to trust an ordering to know whether to keep it; only the hub's own lock can answer, since a hub reads open until the moment it closes under it. Every other caller is drawing a line that belongs to the turn it is in and correctly ignores the answer. `TestALineSaidOntoAClosedStreamIsHeldInstead` pins the property with the pointer deliberately left on a closed hub."
10+
- "A READING'S LANDING IS A FACT A TEST CAN WAIT ON. `readBeside` (sidecar.go) tells a `besideWatch` carried on the work's context when each reading under it STARTS and when it has LANDED, the second after the settle closes, so whoever the watch lets go of finds the answer there to take. `besideWatch.quiet(ctx)` waits until none is in flight OR until the work those readings belong to is over — bounded by that fact and not by a clock, because `sidecar.end` cancels a reading's window without joining it. It is `desk.settled`'s bargain for the other half of that file: THE RUNNING PRODUCT CARRIES NO WATCH AND NEVER WAITS ON ONE, and `TestTheRunningProductNeverWaitsOnAReadingsLanding` reads the tree with `go/ast` — so it is on the laws gate — and fails the build on either of the two ways the product would reach for one."
11+
- "THREE FIXTURES NO LONGER GUESS AT THE SCHEDULER. A scripted conversation answers in no time and never blocks, so a reading the turn has just started can sit unscheduled behind it for as many rounds as the machine is busy: in every failing run the mark's drawing was still pending at all eight remaining boundaries, the script ended in words, and `markAside.takeAtTheEnd` journalled a drawing nobody could spend. A fixture that opts in with `watchReadings(t, agent)` answers the conversation only once every reading beside its turn has landed (`answerWhenQuiet`, applied by `scriptedCompleter`, `routeCompleter` and `reflexScript` to the request that carries the belt), which is the order a real turn usually has: a real model spends seconds on a step, the readings beside it are small, and a drawing that splits cuts the step it lands in. `racingAgent`, `stewardCheckpointAgent` and the three route tests that held their first answer take it."
12+
- "`routeCompleter.holdUntilRaced`, `awaitRace` AND `stopHolding` ARE DELETED. Holding the conversation until the pre-turn calls had been ENTERED — in a loop that slept a millisecond at a time — said nothing about the verdict having been parsed and settled, which is what the boundary after the first word kept failing to find. Nothing sets a hold count now; the watch is the whole of it."
13+
- "A SCRIPTED ARM ORDERS ITSELF BY A SIGNAL, INCLUDING ITS FIRST TOKEN. `lanestub.Profile.FirstTokenUntil` holds a lane's first word until a channel closes, after it has waited out its TTFT as scripted — `StallUntil`'s stated rule for the one wait `StallUntil` cannot hold, since that one only takes hold after `StallAfter` deltas. It holds the whole reply on the unstreamed path, where the first token and the last arrive together. `TestARescueTheCallCouldNotPayForSaysSoOnTheRow` needed it: lane A was scripted at sixty milliseconds against a ceiling of fifty, and a starved timer spent that ten-millisecond margin, so the first token arrived before the controller acted and the row said nothing had been done at all (`action = \"\"`, once in three hundred runs of #924's own head under twelve busy cores). It now holds A's first word on `theControllersWord`, which closes on any rung of the waiting ladder — the report of either half, a rescue, a model switch — so a regression that acts differently fails on its assertion rather than hanging."
14+
- "`docs/rules/ci.md` SAYS WHAT TO DO WITH A TEST THAT FAILS ONLY BESIDE ANOTHER SUITE — reproduce it under load, wait on the fact through the door that owns it, and never `t.Skip`, a retry or a wider bound. It is on that page because the known-red ledger's rules are, and because nothing may be added to that ledger."
15+
---
16+
17+
Five tests were green alone and red beside another suite, a different one each
18+
run, none of them on the ledger. One was the product's: a memory line a person
19+
is owed could be taken off its queue by a reading that had outlived its turn and
20+
said to a stream that had already closed. The other four were tests that
21+
measured the scheduler and called it the road — a count of the calls a fixture
22+
had answered, a script long enough on a quiet machine, ten milliseconds of wall
23+
clock between a ceiling and a first token. What replaces all three guesses is a
24+
wait on the fact itself.
25+
26+
Still true and not fixed here: a drawing that splits can miss the boundary its
27+
own cut opened, because `act` runs before the settle by #871's ruling and a loop
28+
that reaches `take` in that window finds the reading pending — it pays another
29+
generation for the drawing, or drops it when that generation is the turn's last.
30+
The same gap is there when the drawing lands between a boundary and the next
31+
request, where `Agent.cutGeneration` finds nothing in flight to cut.

‎docs/rules/ci.md‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -157,6 +157,34 @@ run on 2026-08-31; #408 took the first five out. What it holds now, and which
157157
entries are Linux-only, is written in the file's own comments and nowhere else —
158158
`CLAUDE.md` no longer copies it, and neither does this page.
159159

160+
## A test that fails only beside another suite
161+
162+
It is a bug report, and it never goes on the ledger above. Reproduce it under
163+
load — a focused `-count=50` at `GOMAXPROCS=2` with a few `yes > /dev/null`
164+
beside it, **on Spark and never on the laptop**, since the point is to starve
165+
the machine — and fix the CAUSE, which is almost always that the test measured
166+
the SCHEDULER and called it the road:
167+
168+
- **Wait on the fact, never on a figure.** A fixed number of scripted rounds, a
169+
margin between two wall-clock numbers, slack "to one short of the next mark" —
170+
each of those is green on a quiet machine and red beside another suite, and
171+
widening one is not a fix.
172+
- **A reading beside the work is waited on through the door that owns it.** A
173+
scripted conversation answers in no time and never blocks, so a reading the
174+
turn has just started may not have been scheduled at all; `internal/session`'s
175+
`besideWatch` (sidecar.go) is told by `readBeside` itself when each reading
176+
under a turn's context starts and lands, and a fixture that opts in
177+
(`watchReadings`) answers only once none is in flight. The product carries no
178+
watch, and a law test fails the build if it ever does.
179+
- **A scripted arm orders itself by a signal.** `internal/lane/lanestub`'s
180+
`StallUntil` holds a stalled answer on a channel and `FirstTokenUntil` holds
181+
the first word the same way, both after the scripted wait is spent — so a
182+
primary that must answer *after* the caller has acted says exactly that,
183+
rather than being scripted a few milliseconds past a bound a busy machine
184+
eats.
185+
- **Never `t.Skip`, never a retry, never a wider timeout.** A bound is for
186+
failing honestly when the fact never arrives, not for passing.
187+
160188
## What blocks a merge
161189

162190
Required today: **`check`** on `dev`, **`cross build`** on `staging` and `main`.

‎internal/lane/lanestub/lanestub.go‎

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -117,6 +117,23 @@ type Profile struct {
117117
// Held on a channel the test never closes in time, the primary CANNOT
118118
// finish first, and the assertion is about the rule again.
119119
StallUntil <-chan struct{}
120+
// FirstTokenUntil is THE SAME RULE FOR THE FIRST TOKEN, which StallUntil
121+
// cannot hold: it only takes hold after StallAfter deltas, so a lane whose
122+
// very first word must land after something else happens has no way to say
123+
// so. With this set, the lane waits out its TTFT as scripted and then waits
124+
// for the channel to close as well — the arm answers no sooner than both.
125+
//
126+
// A primary scripted to answer "a little after" a bound the caller enforces
127+
// is the shape this is for. Sixty milliseconds against a ceiling of fifty
128+
// rests on ten milliseconds of wall clock, which a starved machine eats:
129+
// the caller's timer fires late, the first token arrives first, and the test
130+
// asserts the slack between two figures instead of the rule it was written
131+
// for. Held on a signal the caller itself raises, the order cannot invert.
132+
//
133+
// It holds the whole answer on the unstreamed path, where the first token
134+
// and the last arrive together — a knob that silently did nothing on one of
135+
// the two shapes would be a fixture lying about what it staged.
136+
FirstTokenUntil <-chan struct{}
120137
// FailWith is an HTTP status this lane answers with instead of streaming.
121138
// Zero serves normally.
122139
FailWith int
@@ -1198,6 +1215,10 @@ func (s *Server) serveStream(w http.ResponseWriter, r *http.Request, clock Clock
11981215
s.cancelled(lane.Name)
11991216
return
12001217
}
1218+
if !lane.heldWord(ctx) {
1219+
s.cancelled(lane.Name)
1220+
return
1221+
}
12011222

12021223
total := lane.Tokens
12031224
if total <= 0 {
@@ -1384,6 +1405,10 @@ func (s *Server) serveWhole(w http.ResponseWriter, r *http.Request, clock Clock,
13841405
s.cancelled(lane.Name)
13851406
return
13861407
}
1408+
if !lane.heldWord(r.Context()) {
1409+
s.cancelled(lane.Name)
1410+
return
1411+
}
13871412
var answer strings.Builder
13881413
if lane.Answer != "" {
13891414
answer.WriteString(lane.Answer)
@@ -1413,6 +1438,22 @@ func (s *Server) serveWhole(w http.ResponseWriter, r *http.Request, clock Clock,
14131438
})
14141439
}
14151440

1441+
// heldWord waits for the signal [Profile.FirstTokenUntil] names, and reports
1442+
// whether the lane may now speak: false is a client that went away while the
1443+
// lane was still holding its first word, which is counted like any other
1444+
// cancellation.
1445+
func (lane Lane) heldWord(ctx context.Context) bool {
1446+
if lane.FirstTokenUntil == nil {
1447+
return true
1448+
}
1449+
select {
1450+
case <-lane.FirstTokenUntil:
1451+
return true
1452+
case <-ctx.Done():
1453+
return false
1454+
}
1455+
}
1456+
14161457
// cancelled records a client that walked away. It is counted per lane because
14171458
// the question a hedge test asks is not "was anything cancelled" but "was the
14181459
// LOSER cancelled", and on about twenty lanes that is what stops the bill.

‎internal/lane/lanestub/lanestub_test.go‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -397,6 +397,51 @@ func TestAScriptedMinuteCostsNoWallTime(t *testing.T) {
397397
}
398398
}
399399

400+
// TestALaneHoldsItsFirstWordForASignal is [Profile.FirstTokenUntil], and what
401+
// it asserts is an ORDER rather than an interval: the answer cannot arrive
402+
// while the signal is open, however fast the machine running it is. The scripted
403+
// five seconds are spent on the fast clock before the hold is even reached, so
404+
// what keeps the stream from finishing is the channel and nothing else.
405+
func TestALaneHoldsItsFirstWordForASignal(t *testing.T) {
406+
hold := make(chan struct{})
407+
server := New(model, Lane{Name: "Held", Profile: Profile{
408+
TTFT: 5 * time.Second, Rate: 1000, Tokens: 4, FirstTokenUntil: hold,
409+
}})
410+
defer server.Close()
411+
clock := NewFast(time.Date(2026, 9, 11, 12, 0, 0, 0, time.UTC))
412+
server.SetClock(clock)
413+
414+
done := make(chan []string, 1)
415+
go func() {
416+
done <- ask(t, context.Background(), server, `{"model":"`+model+`","stream":true,"messages":[]}`)
417+
}()
418+
deadline := time.Now().Add(5 * time.Second)
419+
for server.Requests("Held") == 0 && time.Now().Before(deadline) {
420+
time.Sleep(time.Millisecond)
421+
}
422+
if server.Requests("Held") == 0 {
423+
t.Fatal("the request never reached the lane")
424+
}
425+
426+
select {
427+
case <-done:
428+
t.Fatal("the stream finished while its first word was still held")
429+
case <-time.After(50 * time.Millisecond):
430+
}
431+
close(hold)
432+
433+
lines := <-done
434+
if lines[len(lines)-1] != "data: [DONE]" {
435+
t.Fatalf("the released stream did not finish: %d lines", len(lines))
436+
}
437+
// A HOLD SPENDS NO SCRIPTED TIME: the ledger is still the sum of the
438+
// durations the script named, so a scenario's own arithmetic is untouched
439+
// by how long the signal took to arrive.
440+
if clock.Elapsed() > 6*time.Second {
441+
t.Fatalf("the hold spent %s of scripted time", clock.Elapsed()-5*time.Second)
442+
}
443+
}
444+
400445
// TestALaneMayRefuse stages the failure the endpoint ladder reads.
401446
func TestALaneMayRefuse(t *testing.T) {
402447
server := New(model, Lane{Name: "Busy", Profile: Profile{FailWith: http.StatusTooManyRequests}})

‎internal/provider/hedge_refusal_test.go‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,10 +14,19 @@ import (
1414
// `budget`, which named a rolling process-wide allowance and therefore said
1515
// "some other request spent this one's rescue"; [planCannotPay] names the only
1616
// rail there is.
17+
//
18+
// A ANSWERS AFTER THE CONTROLLER HAS SPOKEN, by the order of events. Scripted at
19+
// sixty milliseconds against a ceiling of fifty, the claim rested on ten
20+
// milliseconds of wall clock: a loaded machine spends that before the ceiling's
21+
// timer fires, the first token arrives first, and the row then says nothing was
22+
// done at all — `action = ""`, measured one run in three hundred on this
23+
// commit's own parent under twelve busy cores. So the primary holds its first
24+
// word on the controller's own word ([theControllersWord]).
1725
func TestARescueTheCallCouldNotPayForSaysSoOnTheRow(t *testing.T) {
1826
read := loggingTo(t)
27+
spoken := theControllersWord(t)
1928
rig := newLaneRig(t, "row/budget-refusal",
20-
lanestub.Lane{Name: "A", Profile: lanestub.Profile{TTFT: 60 * time.Millisecond, Rate: 2000, Tokens: 24}},
29+
lanestub.Lane{Name: "A", Profile: lanestub.Profile{TTFT: 60 * time.Millisecond, Rate: 2000, Tokens: 24, FirstTokenUntil: spoken}},
2130
lanestub.Lane{Name: "B", Profile: lanestub.Profile{TTFT: 5 * time.Millisecond, Rate: 2000, Tokens: 24}},
2231
)
2332
rig.believes("A", 20, 2000)

‎internal/provider/hedge_test.go‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -192,6 +192,39 @@ func (r *laneRig) patience(_ *testing.T, ceiling time.Duration) {
192192
r.ceiling.Store(int64(ceiling))
193193
}
194194

195+
// theControllersWord is a signal that closes once the waiting controller has
196+
// spoken: said the wait out loud, reported the pace, or started a rescue. A
197+
// lane given it as its [lanestub.Profile.FirstTokenUntil] answers no sooner.
198+
//
199+
// IT IS THE STUB'S SIGNAL RULE APPLIED TO A FIRST TOKEN, and it exists because
200+
// a primary scripted to answer a little after the ceiling asserts nothing but
201+
// the slack between two wall-clock figures, which a starved machine eats: the
202+
// ceiling's timer fires late, the first token arrives first, and the row says
203+
// nothing was done at all. Every rung of the ladder releases it, not only the
204+
// one a given test is about, so a regression that acts differently fails on
205+
// its assertion rather than hanging here.
206+
func theControllersWord(t *testing.T) <-chan struct{} {
207+
t.Helper()
208+
spoken := make(chan struct{})
209+
var once sync.Once
210+
// The reader that was there is taken FIRST and only then chained, because a
211+
// closure that assigns the handler it chains to is a write racing its own
212+
// reads: [OnPhase] has installed it by the time the assignment happens, and
213+
// a phase posted in that window reads the variable from another goroutine.
214+
previous := OnPhase(nil)
215+
OnPhase(func(news PhaseNews) {
216+
switch news.Phase {
217+
case PhaseAllSlow, PhaseBelowPace, PhaseSwitching, PhaseSwitchingModel:
218+
once.Do(func() { close(spoken) })
219+
}
220+
if previous != nil {
221+
previous(news)
222+
}
223+
})
224+
t.Cleanup(func() { OnPhase(previous) })
225+
return spoken
226+
}
227+
195228
// rigScale is how much shorter every bound a scenario is judged against is than
196229
// the one a person is really given.
197230
//

‎internal/session/agent.go‎

Lines changed: 12 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1597,7 +1597,7 @@ func (a *Agent) startTurnLocked(ctx context.Context, user userMessage, watcher *
15971597
//
15981598
// IT IS STARTED UNDER THIS LOCK for the title's reason as well: two Submits
15991599
// racing to be the first must not each buy a route.
1600-
a.startRecallLocked(turnCtx, hub, user.text())
1600+
a.startRecallLocked(turnCtx, user.text())
16011601
// THEIR NEXT WORDS ARE WHAT CHANGED. A generation Interrupt minted waits
16021602
// here for the sentence that follows Esc, and that sentence is the one
16031603
// decision the leftover handlers and this turn's opening share.
@@ -3542,18 +3542,25 @@ func (h *eventHub) adopt(stream *eventStream) {
35423542
// so every subscriber sees the same events in the same order, and a subscriber
35433543
// arriving mid-fan-out lands cleanly before or after this event rather than
35443544
// inside it.
3545-
func (h *eventHub) send(event Event) {
3545+
//
3546+
// IT REPORTS WHETHER THE EVENT LANDED, because a caller holding something a
3547+
// person is owed must be able to keep it rather than trust an ordering. Only
3548+
// this lock can answer that: a hub reads open until the moment it closes under
3549+
// it, so anything a caller checked beforehand is already a guess. Almost every
3550+
// caller is drawing a line that belongs to the turn it is in and correctly
3551+
// ignores the answer; [Agent.sayMemory] is the one that holds a line over.
3552+
func (h *eventHub) send(event Event) (landed bool) {
35463553
// NOBODY WATCHING IS AN ORDINARY CASE. A completion driven without a turn
35473554
// around it — a test of the retry loop, a future headless caller — has no
35483555
// hub, and a line nobody can read is a line worth not drawing rather than a
35493556
// panic in the middle of a request.
35503557
if h == nil {
3551-
return
3558+
return false
35523559
}
35533560
h.mu.Lock()
35543561
defer h.mu.Unlock()
35553562
if h.closed {
3556-
return
3563+
return false
35573564
}
35583565
// The backlog is written BEFORE the fan-out and under the same lock, so what
35593566
// the next attacher is handed is exactly what the subscribers already have —
@@ -3571,6 +3578,7 @@ func (h *eventHub) send(event Event) {
35713578
for _, stream := range h.subscribers {
35723579
stream.send(event)
35733580
}
3581+
return true
35743582
}
35753583

35763584
// foldedLocked settles the accumulated run back into the backlog entry it

0 commit comments

Comments
 (0)