From 1d7124e388a9cca3c0fb09c438b63586dc0912a0 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 12:06:35 +0100 Subject: [PATCH] DLV-FB1 (indexer's side): a FORGE message of a kind the indexer does not take is acknowledged and passed over The forge bridge now says a branch's end as branch-deleted (design 51); the indexer terminated it as unreadable. It acknowledges it, writes it to heard as ignored, and the two handle tests are one table (mesh-delivery is over its unit budget). Issue 1966. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig --- .../cmd/mesh-delivery-indexer/main.go | 24 +++++++++++ .../cmd/mesh-delivery-indexer/main_test.go | 40 ++++++++++++------- .../internal/index/heard.go | 3 ++ sdk/views/views_test.go | 4 +- 4 files changed, 56 insertions(+), 15 deletions(-) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index 1f0144bf..a8c1f92d 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -10,6 +10,7 @@ package main import ( "context" + "encoding/json" "errors" "fmt" "io" @@ -171,6 +172,13 @@ func handle(ctx context.Context, m delivered, ix *index.Indexer, heard index.Sto logger.Printf("%s: sequence %d is not written to the heard state: %v", h.Subject, h.Sequence, err) } }() + if kind, known := forgeKind(m.Data()); !known { + // A kind the indexer does not take (the bridge's branch-deleted, design 51) is acknowledged and passed over, never + // terminated as unreadable: a new kind on FORGE is no fault of the message. + h.Kind, h.Outcome, h.Words = kind, index.HeardIgnored, "a "+kind+" message, which the indexer does not take" + _ = m.Ack() + return + } f, err := delivery.DecodeForge(m.Data()) if err != nil { logger.Printf("%s: not a forge message, dropped: %v", m.Subject(), err) @@ -203,6 +211,22 @@ func handle(ctx context.Context, m delivered, ix *index.Indexer, heard index.Sto } } +// forgeKind is the kind a FORGE message names, and whether it is one the indexer takes; a body naming no kind is taken, +// so DecodeForge says why it is unreadable. +func forgeKind(data []byte) (string, bool) { + var probe struct { + Kind string `json:"kind"` + } + if json.Unmarshal(data, &probe) != nil || probe.Kind == "" { + return "", true + } + switch probe.Kind { + case delivery.ForgePull, delivery.ForgeMerged, delivery.ForgeRefused, delivery.ForgeOpen: + return probe.Kind, true + } + return probe.Kind, false +} + func opOf(e jetstream.KeyValueEntry) bool { return e.Operation() == jetstream.KeyValuePut } diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go index eb70b62f..4d9c075b 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go @@ -71,6 +71,32 @@ func (m *memStore) Delete(_ context.Context, k string, _ uint64) error { delete( const head = "0123456789abcdef0123456789abcdef01234567" func TestAMessageIsIndexedAndAcked(t *testing.T) { // 1238/AC2 (unit part) + t.Run("not a forge message: terminated and logged", func(t *testing.T) { // 1238 (never redelivered) + var logs bytes.Buffer + st := &memStore{vals: map[string][]byte{}} + m := &fakeMsg{subject: "mesh.stream.mesh-delivery-forge-bridge.event.forge.x.y.pull", data: []byte(`{"v":1,"kind":"pull"}`)} + heard := &memStore{vals: map[string][]byte{}} + handle(context.Background(), m, &index.Indexer{Store: st, Builds: index.NewBuilds()}, heard, log.New(&logs, "", 0)) + if m.termd != 1 || m.acked != 0 || len(st.vals) != 0 || !strings.Contains(logs.String(), m.subject) { + t.Errorf("terminated %d acked %d keys %d log %q", m.termd, m.acked, len(st.vals), logs.String()) + } + if h := heardOf(t, heard); h.Outcome != index.HeardDropped || h.Sequence != 41 || h.Words == "" { + t.Errorf("heard = %+v", h) + } + }) + t.Run("a kind it does not take (branch-deleted): acknowledged and passed over", func(t *testing.T) { // design 51, DLV-FB1 + st := &memStore{vals: map[string][]byte{}} + m := &fakeMsg{subject: "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-sdk.branch-deleted", + data: []byte(`{"v":1,"kind":"branch-deleted","repository":"novox/mesh-sdk","branch":"feat/x","head":"` + head + `"}`)} + heard := &memStore{vals: map[string][]byte{}} + handle(context.Background(), m, &index.Indexer{Store: st, Builds: index.NewBuilds()}, heard, log.New(&bytes.Buffer{}, "", 0)) + if m.termd != 0 || m.acked != 1 || len(st.vals) != 0 { + t.Errorf("terminated %d acked %d keys %d", m.termd, m.acked, len(st.vals)) + } + if h := heardOf(t, heard); h.Outcome != index.HeardIgnored || h.Kind != "branch-deleted" { + t.Errorf("heard = %+v", h) + } + }) st := &memStore{vals: map[string][]byte{}} ix := &index.Indexer{Store: st, Builds: index.NewBuilds()} data, _ := delivery.EncodeForge(delivery.Forge{Kind: delivery.ForgePull, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: head, @@ -90,20 +116,6 @@ func TestAMessageIsIndexedAndAcked(t *testing.T) { // 1238/AC2 (unit part) } } -func TestAMessageThatIsNoForgeMessageIsTerminatedAndLogged(t *testing.T) { // 1238 (never redelivered) - var logs bytes.Buffer - st := &memStore{vals: map[string][]byte{}} - m := &fakeMsg{subject: "mesh.stream.mesh-delivery-forge-bridge.event.forge.x.y.pull", data: []byte(`{"v":1,"kind":"pull"}`)} - heard := &memStore{vals: map[string][]byte{}} - handle(context.Background(), m, &index.Indexer{Store: st, Builds: index.NewBuilds()}, heard, log.New(&logs, "", 0)) - if m.termd != 1 || m.acked != 0 || len(st.vals) != 0 || !strings.Contains(logs.String(), m.subject) { - t.Errorf("terminated %d acked %d keys %d log %q", m.termd, m.acked, len(st.vals), logs.String()) - } - if h := heardOf(t, heard); h.Outcome != index.HeardDropped || h.Sequence != 41 || h.Words == "" { - t.Errorf("heard = %+v", h) - } -} - func TestTheSourceMainKeyIsTakenAtItsRevision(t *testing.T) { // 1344/AC1 (the watch's side: key, revision, value) var logs bytes.Buffer logger := log.New(&logs, "", 0) diff --git a/modules/mesh-delivery-indexer/internal/index/heard.go b/modules/mesh-delivery-indexer/internal/index/heard.go index 0c662255..86ae640b 100644 --- a/modules/mesh-delivery-indexer/internal/index/heard.go +++ b/modules/mesh-delivery-indexer/internal/index/heard.go @@ -21,6 +21,9 @@ const ( HeardApplied = "applied" HeardLeft = "left for redelivery" HeardDropped = "dropped" + // HeardIgnored is a message of a kind the indexer does not take (a branch-deleted one, design 51): acknowledged and + // passed over, as every existing consumer of FORGE passes over a kind it does not know. + HeardIgnored = "ignored" ) // Heard is one FORGE message as the indexer handled it. diff --git a/sdk/views/views_test.go b/sdk/views/views_test.go index 47894933..da65aada 100644 --- a/sdk/views/views_test.go +++ b/sdk/views/views_test.go @@ -114,7 +114,9 @@ func TestViews(t *testing.T) { Raise: func(s Status) { said = append(said, "raised "+s.Tool) }, Clear: func(s Status) { said = append(said, "cleared") }} return f, &said } - answer := func(context.Context, string) (Answer, error) { return FirstView("o", "modules", values, marks, "", now()) } + answer := func(context.Context, string) (Answer, error) { + return FirstView("o", "modules", values, marks, "", now()) + } ev := func(rev uint64, key, value string) Event { e := Event{V: Version, Owner: "o", State: "modules", Key: key, Revision: rev, Created: made} if value == "" {