From 17dd41696cdb2a958477f1eb86d929573055828b Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 10 Nov 2026 03:52:04 +0100 Subject: [PATCH] =?UTF-8?q?D4f:=20a=20merged=20message=20still=20unread=20?= =?UTF-8?q?five=20minutes=20on,=20the=20forge=20heard,=20is=20the=20indexe?= =?UTF-8?q?r's=20finding=20call-unanswered=20(issue=201966=20note=20153,?= =?UTF-8?q?=20ADR=200668=20=C2=A73)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A message heard on the forge is the forge bridge heard, the callee of the call: after each one a merge unread since CallUnansweredAfter or longer writes the finding naming the bridge's call, and a merge no longer unread clears it. While nothing is heard on the forge (the bridge's machine away) no finding is written; the merge is read again when the forge is heard. No timer. Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../internal/index/pending.go | 27 ++++++-- .../internal/index/pending_test.go | 38 ++++++++++ .../internal/index/unreadcall.go | 69 +++++++++++++++++++ 3 files changed, 129 insertions(+), 5 deletions(-) create mode 100644 modules/mesh-delivery-indexer/internal/index/unreadcall.go diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index 962b8ff2..238fa5e1 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending.go +++ b/modules/mesh-delivery-indexer/internal/index/pending.go @@ -92,6 +92,9 @@ type Pending struct { // ADR 0664's classes): not lost, it is read again on the next message heard on the forge, at the start and by // indexer_take_in_pending (issue 1966 note 153, ADR 0668 §1). Unread string `json:"unread,omitempty"` + // UnreadSince is when the message was first found unread: past meshconditionsclient.CallUnansweredAfter, with the forge + // heard since, it is the finding call-unanswered (unreadcall.go, ADR 0668 §3). + UnreadSince time.Time `json:"unread-since,omitzero"` } // PendingEntry is one key of the pending state as read: its key, revision, value and the merged message decoded. For a @@ -698,10 +701,17 @@ func (ix *Indexer) UnreadAgain(ctx context.Context) error { if err != nil { return err } - if !slices.ContainsFunc(entries, func(e PendingEntry) bool { return e.Value.Unread != "" }) { - return nil + if slices.ContainsFunc(entries, func(e PendingEntry) bool { return e.Value.Unread != "" }) { + if err := ix.PendingAgain(ctx); err != nil { + return err + } + if entries, err = ReadPending(ctx, ix.Pending, ix.logf); err != nil { + return err + } } - return ix.PendingAgain(ctx) + // The forge was heard, so the bridge that says it is: a read still failing past the bound is a finding (ADR 0668 §3). + ix.unreadFindings(ctx, entries) + return nil } // PendingAgain evaluates pending again after something a fingerprint is computed from changed, a toolchain report @@ -817,9 +827,16 @@ func (ix *Indexer) holdLost(ctx context.Context, ready, lost []PendingEntry, rea } for _, g := range lost { reason, isUnread := reasons[g.Key], unread[g.Key] + now := ix.now().UTC() ix.markSaid(ctx, []PendingEntry{g}, &t, func(p *Pending) { - if p.Lost, p.Unread = reason, ""; isUnread { - p.Lost, p.Unread = "", reason + p.Lost, p.Unread = reason, "" + if !isUnread { + p.UnreadSince = time.Time{} + return + } + p.Lost, p.Unread = "", reason + if p.UnreadSince.IsZero() { + p.UnreadSince = now } }) } diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go index 672aa599..066f9450 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_test.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_test.go @@ -11,6 +11,8 @@ import ( "testing" "time" + mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" @@ -584,6 +586,8 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) {"the bridge's answer", &busretry.Answered{Words: "no merge at that commit"}, false}, } { b = held() + clock := time.Date(2026, 11, 10, 1, 0, 0, 0, time.UTC) + b.ix.Now, b.ix.Findings = func() time.Time { return clock }, newMem() answers := b.ix.Message b.ix.held = nil b.ix.Message = func(context.Context, string, string, int) ([]byte, error) { return nil, c.fault } @@ -594,6 +598,37 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) if len(waits) != 1 || (len(waits[0].Unread) == 1) != c.unread || (len(waits[0].Lost) == 1) == c.unread { t.Fatalf("%s: indexer_pending %+v: want unread %v, lost %v", c.name, waits, c.unread, !c.unread) } + // ADR 0668 §3: unread within five minutes is no finding; past them, with nothing heard on the forge (the bridge's + // machine away, only a toolchain report heard), none either; the next message heard on the forge, the read still + // failing, writes call-unanswered naming the bridge's call; the merge read and taken in clears it. + finding := func() (string, bool) { + keys, _ := b.ix.Findings.Keys(ctx) + for _, k := range keys { + if strings.HasSuffix(k, "."+mcc.KindCallUnanswered) { + v, _, _, _ := b.ix.Findings.Get(ctx, k) + return string(v), true + } + } + return "", false + } + if c.unread { + for _, step := range []struct { + after time.Duration + heard bool + finding bool + }{{time.Minute, true, false}, {5 * time.Minute, false, false}, {0, true, true}} { + clock = clock.Add(step.after) + var err error + if step.heard { + err = b.ix.UnreadAgain(ctx) + } else { + err = b.ix.PendingAgain(ctx) + } + if v, found := finding(); err != nil || found != step.finding || found && !strings.Contains(v, UnreadCallee) { + t.Fatalf("%s, %s later, forge heard %v: finding %v %s (%v), want %v", c.name, step.after, step.heard, found, v, err, step.finding) + } + } + } b.ix.Message = answers if err := b.ix.UnreadAgain(ctx); err != nil { t.Fatal(err) @@ -601,6 +636,9 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) if taken := len(b.keys()) == 0 && b.commit(t, "avahi") == sha('d'); taken != c.unread { t.Errorf("%s, a message heard after: pending %v, avahi at %s: want taken in %v", c.name, b.keys(), b.commit(t, "avahi"), c.unread) } + if v, found := finding(); found { + t.Errorf("%s: the finding stands after the merge was taken in: %s", c.name, v) + } } }) t.Run("T5 two take-ins of one pending revision: the one that loses the claim writes nothing", func(t *testing.T) { diff --git a/modules/mesh-delivery-indexer/internal/index/unreadcall.go b/modules/mesh-delivery-indexer/internal/index/unreadcall.go new file mode 100644 index 00000000..e8d2f1da --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/unreadcall.go @@ -0,0 +1,69 @@ +package index + +import ( + "context" + "strings" + + mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" +) + +// **A merged message unread past five minutes, the forge heard, is a finding** (issue 1966 note 153; novox/hq ADR 0668 §3). +// The indexer's call for a merged message is a kept call: a transport fault leaves the merge unread (Pending.Unread) and +// it is read again after every message heard on the forge (UnreadAgain). A message heard on the forge is the forge bridge +// heard, the callee of that call: so after each one, a merge still unread since meshconditionsclient.CallUnansweredAfter +// or longer is the finding call-unanswered in the indexer's findings, which the condition keeper raises, and a merge no +// longer unread (read, taken in, covered or dropped) clears its finding. While nothing is heard on the forge (the bridge's +// machine away) no finding is written: the merge stays unread and is read again when the forge is heard. No timer. + +// UnreadCallee is the call the finding names: the forge bridge's tool for one merged message. +const UnreadCallee = ForgeBridgeModule + ".bridge_merged_message" + +// unreadCallID is the finding's id for the pending key: the merge it waits on. +func unreadCallID(key string) string { return "merged-message." + key } + +// unreadFindings raises the finding of every merge of entries unread past the bound and clears the finding of every other. +func (ix *Indexer) unreadFindings(ctx context.Context, entries []PendingEntry) { + if ix.Findings == nil { + return + } + unread := map[string]bool{} + for _, e := range entries { + if e.Value.Unread == "" { + continue + } + unread[mcc.Key(mcc.ScopeCall, unreadCallID(e.Key), mcc.KindCallUnanswered)] = true + since := e.Value.UnreadSince + if since.IsZero() || ix.now().Sub(since) < mcc.CallUnansweredAfter { + continue + } + o := mcc.CallUnanswered(unreadCallID(e.Key), UnreadCallee, "", e.Value.Unread, "indexer."+mcc.KindCallUnanswered, since) + o.Needs = TakeInPendingTool + " reads the merge again now; " + PendingTool + " shows it unread." + key := o.Key() + if err := write(ctx, ix.Findings, key, func(value []byte, found bool) ([]byte, bool, error) { + if found { + if f, err := mcc.DecodeFinding(value); err == nil && f.Summary == o.Summary { + return nil, false, nil + } + } + data, err := mcc.EncodeFinding(o.Finding(Module, "", ix.now().UTC())) + return data, false, err + }); err != nil { + ix.logf("take-in: the finding %s is not written: %v", key, err) + } else { + ix.logf("take-in: %s (%s#%d) is unread since %s and the forge is heard: the finding %s", e.Key, e.Forge.Repository, e.Forge.Number, since.Format("2006-01-02T15:04:05Z07:00"), key) + } + } + keys, err := ix.Findings.Keys(ctx) + if err != nil { + return + } + prefix, suffix := mcc.ScopeCall+"."+unreadCallID(""), "."+mcc.KindCallUnanswered + for _, k := range keys { + if !strings.HasPrefix(k, prefix) || !strings.HasSuffix(k, suffix) || unread[k] { + continue + } + if err := write(ctx, ix.Findings, k, func(_ []byte, found bool) ([]byte, bool, error) { return nil, found, nil }); err != nil { + ix.logf("take-in: the finding %s is not cleared: %v", k, err) + } + } +}