novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

D4f: a merged message still unread five minutes on, the forge heard, is the indexer's finding call-unanswered (issue 1966 note 153, ADR 0668 §3)

Browse Source 
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
This commit is contained in:
jschoubben
2026-11-10 03:52:04 +01:00
parent 7b3f213ae8
commit 17dd41696c
3 changed files with 129 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-indexer/internal/index/pending.go
+22 -5
View File
@@ -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
}
})
}
modules/mesh-delivery-indexer/internal/index/pending_test.go
+38
View File
@@ -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) {
modules/mesh-delivery-indexer/internal/index/unreadcall.go
+69
View File
@@ -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)
}
}
}