Skip to content

Commit 078dab7

Browse files
author
gluck
committed
cli: a hop drops frames from the aria it has not claimed yet (#25)
The pager attends a parent; `:fork -- <prompt>` prompts a branch and hops to it. The hop took its generation, dialled the branch (subscribed on accept, streaming at once), waited for the lineage read, and only then swapped the client in claimSubject. notifyHandler's whole fence was the generation, which the new connection already held, so every frame the branch sent in that gap was folded into the PARENT's client, which claimSubject then put on the shelf. Coming back to the parent restored it from the shelf and painted the branch's opening question at the end of the parent's last reply. Caught on a recorded tape: the turn-10 frame at t=66.578, beside form deltas naming the branch, before the client's first read for it at t=66.617. notifyHandler now knows the aria it was dialled for and drops what arrives before that aria is the claimed subject, for aria frames and intrinsic form deltas alike. Nothing is lost: the seed read after the claim covers the same ground, Client.Apply is idempotent, a live frame after a gap trips the desync path, and the intrinsic mirrors are re-seeded whole after the claim. A dropped frame emits hop.early (marks.md). The identity check is added to the generation, not a replacement: a superseded switch's connection stays mute even if a later switch claims its aria. TestHop_AFrameFromTheAriaBeingDialledNeverReachesTheOneBeingLeft drives the exact sequence (parent claimed, hop begun, OpenInquiry in the gap, claim, shelf) and fails on the old fence with "the branch's question "test" landed in the parent's client before the hop claimed the branch". BenchmarkNotifyFrame: ~3.4 us per aria frame, ~1.2 us per form delta, 78 ns per dropped early frame.
1 parent f357911 commit 078dab7

5 files changed

Lines changed: 172 additions & 4 deletions

File tree

‎internal/cli/command.go‎

Lines changed: 27 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -715,7 +715,7 @@ func (in *interactiveInput) retarget(ctx context.Context, id string, ep transpor
715715
go func() { base <- in.divergence(ctx, id, from) }()
716716

717717
dial := mark.Span("hop.dial", "to", id, "gen", gen)
718-
fcli, err := sdk.DialAriaWith(ep, in.notifyHandler(gen), markTap(in.tap))
718+
fcli, err := sdk.DialAriaWith(ep, in.notifyHandler(gen, id), markTap(in.tap))
719719
dial()
720720
// THE ANSWER IS TAKEN BEFORE THE LOCK. Asking the angelus takes in.mu, so
721721
// waiting for it under that lock would park the switch on a goroutine that
@@ -955,8 +955,24 @@ func (in *interactiveInput) seedMetrics(gen uint64) {
955955
}
956956

957957
// notifyHandler folds one connection's frames, and only while that connection
958-
// is the subject.
959-
func (in *interactiveInput) notifyHandler(gen uint64) sdk.NotifyHandler {
958+
// is the subject: dialled for id, under generation gen, AND claimed.
959+
//
960+
// THE GENERATION ALONE WAS NOT A FENCE (issue #25). A switch takes its
961+
// generation before it dials, the connection is subscribed the moment it is
962+
// accepted, and the client is swapped only later, in claimSubject, after the
963+
// lineage read returns. Every frame the new aria sent in between passed the
964+
// generation check and was folded into the client of the aria being LEFT,
965+
// which claimSubject then put on the shelf. A branch's opening question,
966+
// prompted by `:fork -- <prompt>` a moment before the hop, landed in the
967+
// parent's window that way and was painted at the end of the parent's last
968+
// reply when the reader came back.
969+
//
970+
// So the handler knows which aria it was dialled for and drops what arrives
971+
// before that aria is the claimed subject. Nothing is lost by it: the seed
972+
// read that follows the claim covers the same ground, Client.Apply is
973+
// idempotent, a live frame that arrives after a gap trips the desync path and
974+
// re-reads, and the intrinsic mirrors are re-seeded whole after the claim.
975+
func (in *interactiveInput) notifyHandler(gen uint64, id string) sdk.NotifyHandler {
960976
return func(method string, params json.RawMessage) {
961977
if atomic.LoadUint64(&in.subjectGen) != gen {
962978
return
@@ -968,10 +984,18 @@ func (in *interactiveInput) notifyHandler(gen uint64) sdk.NotifyHandler {
968984
// that reads as if it holds the lock throughout. One entrance, one
969985
// lock discipline.
970986
if method == rpc.MethodFormDelta {
987+
if in.currentID() != id {
988+
return
989+
}
971990
in.applyIntrinsicDelta(params)
972991
return
973992
}
974993
in.mu.Lock()
994+
if in.figaroID != id {
995+
in.mu.Unlock()
996+
mark.Mark("hop.early", "to", id, "method", method)
997+
return
998+
}
975999
switch method {
9761000
case rpc.MethodAriaFrame:
9771001
in.turnFrame(params)
Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,93 @@
1+
package cli
2+
3+
import (
4+
"encoding/json"
5+
"sync/atomic"
6+
"testing"
7+
8+
"github.com/jack-work/figaro/api/rpc"
9+
"github.com/jack-work/figaro/internal/livelog/aria"
10+
)
11+
12+
// ISSUE #25, AT THE SEAM WHERE IT LIVED.
13+
//
14+
// The pager attends a parent. `:fork -- test` prompts a branch and then hops
15+
// to it. The hop takes its generation, dials the branch (which is subscribed
16+
// on accept and starts streaming at once), and only then waits for the lineage
17+
// read before claimSubject swaps the client. The branch's opening question,
18+
// an OpenInquiry frame for the turn after the parent's last, arrives in that
19+
// gap. It passed the generation check and was folded into the PARENT's
20+
// client, which claimSubject then shelved; coming back to the parent restored
21+
// it from the shelf and painted the branch's question at the end of the
22+
// parent's reply.
23+
//
24+
// Captured on a real tape (Gluck, 2026-10-10): the turn-10 frame at t=66.578,
25+
// beside form deltas naming the branch, before the client's first read for it
26+
// at t=66.617. This drives the same order without a wire.
27+
func TestHop_AFrameFromTheAriaBeingDialledNeverReachesTheOneBeingLeft(t *testing.T) {
28+
in := fenceSession(t)
29+
const parent, branch = "parentAA", "branchBB"
30+
31+
// The parent is on screen, claimed, with three turns.
32+
gen := atomic.AddUint64(&in.subjectGen, 1)
33+
if _, _, ok := in.claimSubject(gen, parent, "", nil, 0); !ok {
34+
t.Fatal("could not claim the parent")
35+
}
36+
in.lt.apply(aria.Page{Parts: partsFor(1, 3, "PARENT")})
37+
38+
// The hop to the branch begins: a new generation, a connection dialled for
39+
// the branch with its handler, and no claim yet.
40+
gen = atomic.AddUint64(&in.subjectGen, 1)
41+
handler := in.notifyHandler(gen, branch)
42+
43+
// The branch opens its first turn: the question alone, as
44+
// Server.OpenInquiry broadcasts it.
45+
frame, err := json.Marshal(aria.Page{Parts: []aria.TurnPart{{Turn: aria.Turn{ID: 4, Inquiry: "test"}}}})
46+
if err != nil {
47+
t.Fatal(err)
48+
}
49+
handler(rpc.MethodAriaFrame, frame)
50+
51+
if inq, ok := in.lt.client.InquiryOf(4); ok {
52+
t.Fatalf("the branch's question %q landed in the parent's client before the hop claimed the branch", inq.Text)
53+
}
54+
if turn, _, ok := in.lt.client.OpenKey(); ok && turn == 4 {
55+
t.Fatal("the parent's client holds the branch's turn 4 open")
56+
}
57+
58+
// The claim: the parent goes on the shelf, and must go there clean.
59+
if _, _, ok := in.claimSubject(gen, branch, parent, nil, 3); !ok {
60+
t.Fatal("could not claim the branch")
61+
}
62+
shelved := in.parked[parent]
63+
if shelved == nil {
64+
t.Fatal("the parent was not shelved")
65+
}
66+
if _, ok := shelved.client.InquiryOf(4); ok {
67+
t.Fatal("the parent went onto the shelf carrying the branch's question: the hop back will paint it")
68+
}
69+
70+
// After the claim the same frame is the branch's own, and is folded.
71+
handler(rpc.MethodAriaFrame, frame)
72+
if _, ok := in.lt.client.InquiryOf(4); !ok {
73+
t.Fatal("after the claim the branch's own question was dropped: the fence is too wide")
74+
}
75+
}
76+
77+
// A connection whose switch was SUPERSEDED stays mute even if its aria is
78+
// later claimed by another switch: the generation still fences it. The id
79+
// check is added to the generation, it does not replace it.
80+
func TestHop_TheIdentityFenceDoesNotReplaceTheGeneration(t *testing.T) {
81+
in := fenceSession(t)
82+
stale := atomic.AddUint64(&in.subjectGen, 1)
83+
old := in.notifyHandler(stale, "ariaCCCC")
84+
fresh := atomic.AddUint64(&in.subjectGen, 1)
85+
if _, _, ok := in.claimSubject(fresh, "ariaCCCC", "", nil, 0); !ok {
86+
t.Fatal("could not claim")
87+
}
88+
frame, _ := json.Marshal(aria.Page{Parts: partsFor(1, 1, "STALE")})
89+
old(rpc.MethodAriaFrame, frame)
90+
if in.lt.client.Count() != 0 {
91+
t.Fatal("a connection from a superseded switch folded a frame")
92+
}
93+
}

‎internal/cli/hop_idle_reads_test.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -165,7 +165,7 @@ func TestHop_AGapFromTheStreamIsChasedWithoutAKey(t *testing.T) {
165165
in.lt.apply(aria.Page{Parts: partsFor(1, 3, "PARENT")})
166166
in.lt.tr.buildIndex()
167167

168-
handler := in.notifyHandler(in.subjectGeneration())
168+
handler := in.notifyHandler(in.subjectGeneration(), in.currentID())
169169
frame, err := json.Marshal(aria.Page{
170170
Parts: partsFor(9, 1, "LATER"),
171171
More: aria.More{Before: true},

‎internal/cli/notify_bench_test.go‎

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
package cli
2+
3+
import (
4+
"encoding/json"
5+
"sync/atomic"
6+
"testing"
7+
8+
"github.com/jack-work/figaro/api/rpc"
9+
"github.com/jack-work/figaro/internal/livelog/aria"
10+
)
11+
12+
// THE FENCE IS ON THE HOTTEST PATH A PAGER HAS: every frame of every stream
13+
// passes through notifyHandler. This measures one frame through it on a
14+
// claimed subject: an aria page (the identity check is a string compare under
15+
// a lock the handler already takes) and an intrinsic form delta (where the
16+
// check takes the lock once more). And the dropped case, which is what a frame
17+
// costs when it arrives during a hop.
18+
func BenchmarkNotifyFrame(b *testing.B) {
19+
in := fenceSession(b)
20+
in.intrinsics = newIntrinsicMirrors() // as runsession wires a real session
21+
const id = "ariaAAAA"
22+
gen := atomic.AddUint64(&in.subjectGen, 1)
23+
if _, _, ok := in.claimSubject(gen, id, "", nil, 0); !ok {
24+
b.Fatal("could not claim")
25+
}
26+
in.lt.apply(aria.Page{Parts: partsFor(1, 20, "T")})
27+
page, _ := json.Marshal(aria.Page{Parts: partsFor(20, 1, "T")})
28+
delta, _ := json.Marshal(rpc.FormDelta{Intrinsic: "runtime"})
29+
30+
handler := in.notifyHandler(gen, id)
31+
b.Run("frame=aria", func(b *testing.B) {
32+
b.ReportAllocs()
33+
for b.Loop() {
34+
handler(rpc.MethodAriaFrame, page)
35+
}
36+
})
37+
b.Run("frame=form-delta", func(b *testing.B) {
38+
b.ReportAllocs()
39+
for b.Loop() {
40+
handler(rpc.MethodFormDelta, delta)
41+
}
42+
})
43+
early := in.notifyHandler(gen, "ariaBBBB")
44+
b.Run("frame=aria-before-claim", func(b *testing.B) {
45+
b.ReportAllocs()
46+
for b.Loop() {
47+
early(rpc.MethodAriaFrame, page)
48+
}
49+
})
50+
}

‎skills/figaro/contributing/marks.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ as a float.
1717
| `hop` | cli | `from`, `to`, `gen`, `ms` | one subject change, keystroke to seeded window; the phases below add up to it |
1818
| `hop.dial` | cli | `to`, `gen`, `ms` | connecting to the new aria |
1919
| `hop.read` | cli | `kind`, `ms`, `parts`, `err` | one read on the pager's behalf: `seed` (retarget), `enter` (cold pager), `catchup`, `history` (scrolling up), `page` (a jump) |
20+
| `hop.early` | cli | `to`, `method` | a frame from the aria being dialled arrived before the hop claimed it, and was dropped (issue #25: it used to land in the aria being left); the seed read covers it |
2021
| `wire.rx`, `wire.tx` | cli | `conn`, `bytes` | bytes on one aria connection, as they pass |
2122
| `aria.frame` | cli | `bytes`, `parts`, `live`, `deltas` | one `figaro.aria` notification arrived |
2223
| `frame` | cli | `aria` or `view`, `bytes`, `full`, `content`, `tail`, `rows` | the terminal was written: a painted frame. `content` says the transcript had rows; `tail` says the view was following |

0 commit comments

Comments
 (0)