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
This commit is contained in:
2026-11-04 12:22:26 +01:00
parent 9d79379c5f
commit 1d7124e388
4 changed files with 56 additions and 15 deletions
@@ -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
}
@@ -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)
@@ -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.
+3 -1
View File
@@ -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 == "" {