diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go index e836b4e57..651af2836 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go @@ -98,6 +98,7 @@ func main() { func run(h *Holder, l *listening) { <-stdio.Served() // Serve first: the state is reached through it, and an event is heard from as soon as it serves switchToRecords(h, l) + statePins(h, l) go h.Work() for _, pattern := range followed { pattern := pattern @@ -130,6 +131,20 @@ func switchToRecords(h *Holder, l *listening) { } } +// statePins says every pin the records hold once at each start (design 47 G §579), retrying until it can. +func statePins(h *Holder, l *listening) { + for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { + n, err := h.Records.StatePins() + if err == nil { + logf("[mesh-delivery] pins stated: %d", n) + return + } + l.set("not yet: " + err.Error()) + logf("[mesh-delivery] the pins cannot be stated yet (%v); asking again in %s", err, wait) + time.Sleep(wait) + } +} + // handle takes one event, then asks for a look at every step the event makes possible (Work, React). An event that cannot be read is // said and acknowledged: read again it would fail again. func handle(h *Holder, e stdio.Envelope) error { diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/recordevents.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/recordevents.go index 0551093e7..a92f4845f 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/recordevents.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/recordevents.go @@ -255,3 +255,25 @@ func (rs *Records) sayEnded(r *Record) error { return rs.say(meshdelivery.SubjectRecordEnded, recordEnded{Record: r.Instance, Milestone: r.Milestone, ForgeMade: r.ForgeMade, Away: r.Ended.Away, At: r.Ended.At}, "record-ended."+r.Instance) } + +// StatePins says `pinned` again for every revision each record it keeps pins, then once `pins-stated` (design 47 G §579): the +// indexer's consumer of the seat's events hears pins said before it existed or older than the stream keeps, and the set form +// absorbs the repeats. Each repeat has an id of its own, since the bus keeps one message under one id inside its duplicate window. +func (rs *Records) StatePins() (int, error) { + kept, err := rs.Kept() + if err != nil { + return 0, err + } + at := rs.Now().UTC() + stated := 0 + for _, r := range kept { + for _, p := range sortedPins(r) { + id := "pinned." + r.Instance + "." + p.Revision + ".stated." + strconv.FormatInt(at.UnixNano(), 10) + if err := rs.say(meshdelivery.SubjectPinned, pinsOf(r, []PinnedRevision{p}, ""), id); err != nil { + return stated, err + } + stated++ + } + } + return stated, rs.say(meshdelivery.SubjectPinsStated, meshdelivery.PinsStatedEvent{At: at}, "pins-stated."+strconv.FormatInt(at.UnixNano(), 10)) +} diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/sourcetools_test.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/sourcetools_test.go index a842f05ea..b5c142d9c 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/sourcetools_test.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/sourcetools_test.go @@ -58,3 +58,29 @@ func TestTheSourceToolsReadEveryBranchAndSayItAgainAfterTheLog(t *testing.T) { t.Fatalf("log %+v events %+v", acts.took, events) } } + +// Design 47 G §579: at its start the gate says `pinned` again for every revision each record pins, each under an id of its own, +// then `pins-stated` once. +func TestTheGateStatesItsPinsThenSaysPinsStated(t *testing.T) { + at := time.Date(2026, 11, 12, 4, 0, 0, 0, time.UTC) + type said struct{ subject, id string } + var events []said + rs := &Records{Ledger: newLedgerFake(t), Now: func() time.Time { return at }, + Say: func(subject string, body any, id string) error { + events = append(events, said{subject, id}) + return nil + }} + r, _, err := rs.Open("47/G", map[string]string{"novox/app@1": "c1"}, false) + if err != nil { + t.Fatal(err) + } + if err := rs.Pin(r, HalfCatalog, 3); err != nil { + t.Fatal(err) + } + events = nil + n, err := rs.StatePins() + if err != nil || n != 1 || len(events) != 2 || !strings.HasSuffix(events[0].subject, ".event.pinned") || + !strings.Contains(events[0].id, ".stated.") || !strings.HasSuffix(events[1].subject, ".event.pins-stated") { + t.Fatalf("%d %v %+v", n, err, events) + } +}