From d6b6f24e98d2fd2c56df6080f08bd07957a293c0 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 23:11:53 +0100 Subject: [PATCH] =?UTF-8?q?forge=20bridge:=20a=20member's=20merged=20messa?= =?UTF-8?q?ge=20names=20its=20group,=20and=20the=20group's=20end=20is=20sa?= =?UTF-8?q?id=20once=20(ADR=200633=20=C2=A74,=20issue=201585)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The group is read from the gate's merges state the bridge already reads; the end comes only once every approved member is merged and its merged message written, never on a refusal, with a message id the stream's duplicate window holds to one. Claude-Session: https://claude.ai/code/session_01JG73KG7AfSdh6FP8VrBqCn --- .../cmd/mesh-delivery-forge-bridge/groups.go | 215 ++++++++++++++++++ .../mesh-delivery-forge-bridge/groups_test.go | 94 ++++++++ .../cmd/mesh-delivery-forge-bridge/main.go | 26 ++- .../mesh-delivery-forge-bridge/main_test.go | 9 +- 4 files changed, 337 insertions(+), 7 deletions(-) create mode 100644 modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groups.go create mode 100644 modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groups_test.go diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groups.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groups.go new file mode 100644 index 00000000..486990b9 --- /dev/null +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groups.go @@ -0,0 +1,215 @@ +package main + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "log" + "slices" + "strconv" + "strings" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// A delivery group's merges reach the indexer together (novox/hq ADR 0633 §4): the bridge sets the group on each +// member's merged message, from the gate's merges state it reads (ADR 0437 §8), and once every approved member is +// merged and its merged message written on FORGE it says the group's end, forge.group.merged, once. +// +// What it keeps of it is in its state performed, beside the gate's keys it acted on: `published.` for each +// member's merged message written (the hash is the first 16 hex digits of the sha256 of `#@`), +// and `groups.` once the group's end is written, its value the revision of the gate's merges key it was +// worked out from. A group whose end is not written yet is open (bridge_groups reads them). + +// groupsKey is the key of performed that says a group's end is written. +func groupsKey(mergesKey string) string { return "groups." + mergesKey } + +// publishedKey is the key of performed that says a member's merged message is written. +func publishedKey(repository string, number int, merge string) string { + sum := sha256.Sum256([]byte(fmt.Sprintf("%s#%d@%s", strings.ToLower(repository), number, merge))) + return "published." + hex.EncodeToString(sum[:8]) +} + +// groupEndID is the message id of a group's end: the stream's duplicate window drops a second publish of one end. +func groupEndID(group, lastMerge string) string { return "group-end." + group + "." + lastMerge } + +// merges reads one key of the gate's merges state. +type mergesRead func(ctx context.Context, key string) (value []byte, rev uint64, found bool, err error) + +// grouper sets a member's group and says a group's end. +type grouper struct { + merges mergesRead + send func(ctx context.Context, subject string, data []byte, id string) error // writes on FORGE with the message id + done revisions + logger *log.Logger +} + +// sendVia writes on FORGE through pub, with the message id. +func sendVia(pub publisher) func(context.Context, string, []byte, string) error { + return func(ctx context.Context, subject string, data []byte, id string) error { + _, err := pub.Publish(ctx, subject, data, jetstream.WithMsgID(id)) + return err + } +} + +func kvMerges(kv jetstream.KeyValue) mergesRead { + return func(ctx context.Context, key string) ([]byte, uint64, bool, error) { + e, err := kv.Get(ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil, 0, false, nil + } + if err != nil { + return nil, 0, false, err + } + return e.Value(), e.Revision(), true, nil + } +} + +// groupOf is the gate's merge of the group a merged message's branch names, when the message's pull request is one of +// its approved members and the group has more than one: a lone pull request is no group. +func (g *grouper) groupOf(ctx context.Context, f delivery.Forge) (delivery.Merge, uint64, bool) { + if g == nil || g.merges == nil || f.Branch == "" { + return delivery.Merge{}, 0, false + } + key, err := delivery.MergesKey(f.Branch) + if err != nil { + return delivery.Merge{}, 0, false + } + value, rev, found, err := g.merges(ctx, key) + if err != nil { + g.logger.Printf("%s#%d: the gate's merges.%s cannot be read, so its merged message names no group: %v", f.Repository, f.Number, key, err) + return delivery.Merge{}, 0, false + } + if !found { + return delivery.Merge{}, 0, false + } + m, err := delivery.DecodeMerge(value) + if err != nil { + g.logger.Printf("%s#%d: the gate's merges.%s is not a merge, so its merged message names no group: %v", f.Repository, f.Number, key, err) + return delivery.Merge{}, 0, false + } + if _, ok := m.Approved[delivery.ApprovedKey(f.Repository, f.Number)]; !ok || len(m.Approved) < 2 { + return delivery.Merge{}, 0, false + } + return m, rev, true +} + +// membersOf are a merge's approved members with their heads, in the order of their keys. +func membersOf(m delivery.Merge) []delivery.Head { + var out []delivery.Head + for _, k := range slices.Sorted(func(yield func(string) bool) { + for k := range m.Approved { + if !yield(k) { + return + } + } + }) { + at := strings.LastIndex(k, "@") + if at < 0 { + continue + } + n, err := strconv.Atoi(k[at+1:]) + if err != nil { + continue + } + out = append(out, delivery.Head{Repository: k[:at], Number: n, Head: m.Approved[k]}) + } + return out +} + +// member sets the group and its approved members on a merged message's body (ADR 0633 §4); any other body, and a +// merge of no group, is left as it is. +func (g *grouper) member(ctx context.Context, data []byte) ([]byte, delivery.Forge, bool) { + f, err := delivery.DecodeForge(data) + if err != nil || f.Kind != delivery.ForgeMerged { + return data, f, false + } + m, _, ok := g.groupOf(ctx, f) + if !ok { + return data, f, false + } + f.Group, f.Members = m.Group, membersOf(m) + out, err := delivery.EncodeForge(f) + if err != nil { + g.logger.Printf("%s#%d: its merged message cannot name the group %q, written without: %v", f.Repository, f.Number, m.Group, err) + return data, f, false + } + return out, f, true +} + +// published keeps that a member's merged message is written, and says the group's end when it was the last. +func (g *grouper) published(ctx context.Context, f delivery.Forge) { + if err := g.done.Put(ctx, publishedKey(f.Repository, f.Number, f.Merge), 1); err != nil { + g.logger.Printf("%s#%d: that its merged message is written is not kept, so the group's end waits for the next act: %v", f.Repository, f.Number, err) + return + } + key, err := delivery.MergesKey(f.Branch) + if err != nil { + return + } + value, rev, found, err := g.merges(ctx, key) + if err != nil || !found { + return + } + g.ended(ctx, key, rev, value) +} + +// ended says the group's end (forge.group.merged) once every approved member of the gate's merge is merged and its +// merged message written, after the last of them; never on a refusal (ADR 0441 §3: a held group merges on from it). +// It is written once per group: `groups.` in performed says it was. again says it is written again by hand +// (bridge_group_end_again), with the time appended to the message id, so the duplicate window does not drop it. +func (g *grouper) ended(ctx context.Context, key string, rev uint64, value []byte) { + g.end(ctx, key, rev, value, "") +} + +func (g *grouper) end(ctx context.Context, key string, rev uint64, value []byte, again string) bool { + m, err := delivery.DecodeMerge(value) + if err != nil || len(m.Approved) < 2 { + return false + } + if again == "" { + if _, seen, err := g.done.Get(ctx, groupsKey(key)); err == nil && seen { + return false + } + } + end := delivery.ForgeGroupEnd{Group: m.Group} + for _, member := range membersOf(m) { + at := slices.IndexFunc(m.Pulls, func(p delivery.MergePull) bool { + return delivery.SameRepository(p.Repository, member.Repository) && p.Number == member.Number + }) + if at < 0 || m.Pulls[at].State != delivery.MergeMerged || m.Pulls[at].Commit == "" { + return false // not every member is merged: the group stays gathered + } + if _, ok, err := g.done.Get(ctx, publishedKey(member.Repository, member.Number, m.Pulls[at].Commit)); err != nil || !ok { + return false // its merged message is not written yet: said when it is (published) + } + end.Members = append(end.Members, delivery.GroupMerge{Repository: member.Repository, Number: member.Number, Merge: m.Pulls[at].Commit}) + } + // The members in the order the gate merged them, so the last commit names the end. + slices.SortStableFunc(end.Members, func(a, b delivery.GroupMerge) int { + return slices.IndexFunc(m.Pulls, func(p delivery.MergePull) bool { return p.Commit == a.Merge }) - + slices.IndexFunc(m.Pulls, func(p delivery.MergePull) bool { return p.Commit == b.Merge }) + }) + data, err := delivery.EncodeForgeGroupEnd(end) + if err != nil { + g.logger.Printf("the end of the group %q cannot be written: %v", m.Group, err) + return false + } + id := groupEndID(m.Group, end.Members[len(end.Members)-1].Merge) + if again != "" { + id += "." + again + } + if err := g.send(ctx, forgeSubject(delivery.ForgeGroupEndEvent), data, id); err != nil { + g.logger.Printf("the end of the group %q is not written on FORGE, written at the gate's next write or by bridge_group_end_again: %v", m.Group, err) + return false + } + if err := g.done.Put(ctx, groupsKey(key), rev); err != nil { + g.logger.Printf("the end of the group %q is written and not kept: the stream's duplicate window drops a second one: %v", m.Group, err) + } + g.logger.Printf("the end of the group %q is written: %d members merged, the last %s", m.Group, len(end.Members), end.Members[len(end.Members)-1].Merge) + return true +} diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groups_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groups_test.go new file mode 100644 index 00000000..419c259e --- /dev/null +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groups_test.go @@ -0,0 +1,94 @@ +package main + +import ( + "bytes" + "context" + "log" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// idPub records the subjects, bodies and message ids it is handed. +type idPub struct { + subjects, ids []string + data [][]byte +} + +func (p *idPub) send(_ context.Context, subject string, data []byte, id string) error { + p.subjects, p.data, p.ids = append(p.subjects, subject), append(p.data, data), append(p.ids, id) + return nil +} + +func TestAGroupsMembersNameItAndItsEndIsSaidOnce(t *testing.T) { // 1585/T16, T17 (ADR 0633 §4) + ctx := context.Background() + h1, h2 := strings.Repeat("1", 40), strings.Repeat("2", 40) + c1, c2 := strings.Repeat("c", 40), strings.Repeat("d", 40) + merge := func(states ...delivery.MergePull) []byte { + v, err := delivery.EncodeMerge(delivery.Merge{Group: "feat/take-in", Ask: "ask", Pulls: states, + Approved: map[string]string{"novox/mesh-sdk@37": h1, "novox/mesh-delivery@9": h2}}) + if err != nil { + t.Fatal(err) + } + return v + } + sdk := func(state delivery.MergeState, commit, words string) delivery.MergePull { + return delivery.MergePull{Head: delivery.Head{Repository: "novox/mesh-sdk", Number: 37, Head: h1}, State: state, Commit: commit, Words: words} + } + dlv := func(state delivery.MergeState, commit string) delivery.MergePull { + return delivery.MergePull{Head: delivery.Head{Repository: "novox/mesh-delivery", Number: 9, Head: h2}, State: state, Commit: commit} + } + gate := map[string][]byte{} + var logs bytes.Buffer + pub, done := &idPub{}, memRevisions{} + g := &grouper{send: pub.send, done: done, logger: log.New(&logs, "", 0), + merges: func(_ context.Context, key string) ([]byte, uint64, bool, error) { + v, ok := gate[key] + return v, 7, ok, nil + }} + mergedOf := func(repo string, number int, head, commit string) []byte { + data, err := delivery.EncodeForge(delivery.Forge{Kind: delivery.ForgeMerged, Repository: repo, Branch: "feat/take-in", Number: number, Head: head, Merge: commit, Root: "r"}) + if err != nil { + t.Fatal(err) + } + return data + } + // A lone pull request's merged message carries no group. + if _, f, ok := g.member(ctx, mergedOf("novox/hq", 4, h1, c1)); ok || f.Group != "" { + t.Errorf("a lone pull request names the group %q", f.Group) + } + // The first member: its message names the group and both members; the group is not ended. + gate["feat/take-in"] = merge(sdk(delivery.MergeMerged, c1, ""), dlv(delivery.MergeMerging, "")) + data, f, ok := g.member(ctx, mergedOf("novox/mesh-sdk", 37, h1, c1)) + if back, err := delivery.DecodeForge(data); !ok || err != nil || back.Group != "feat/take-in" || len(back.Members) != 2 { + t.Fatalf("a member's message: %+v %v", back, err) + } + g.published(ctx, f) + // A refusal of the other member is no end (ADR 0441 §3: the group merges on from it later). + gate["feat/take-in"] = merge(sdk(delivery.MergeMerged, c1, ""), delivery.MergePull{Head: delivery.Head{Repository: "novox/mesh-delivery", Number: 9, Head: h2}, State: delivery.MergeRefused, Words: "405"}) + g.ended(ctx, "feat/take-in", 8, gate["feat/take-in"]) + // Merged on at the gate before its merged message is written: still no end; then its message ends the group. + gate["feat/take-in"] = merge(sdk(delivery.MergeMerged, c1, ""), dlv(delivery.MergeMerged, c2)) + g.ended(ctx, "feat/take-in", 9, gate["feat/take-in"]) + if len(pub.subjects) != 0 { + t.Fatalf("an end before every member's merged message is written: %v", pub.subjects) + } + _, f, _ = g.member(ctx, mergedOf("novox/mesh-delivery", 9, h2, c2)) + g.published(ctx, f) + g.ended(ctx, "feat/take-in", 10, gate["feat/take-in"]) // a later write of the gate says no second end + if len(pub.subjects) != 1 || pub.subjects[0] != "mesh.stream.mesh-delivery-forge-bridge.event.forge.group.merged" || pub.ids[0] != "group-end.feat/take-in."+c2 { + t.Fatalf("ends %v ids %v, want one forge.group.merged with id group-end.feat/take-in.%s", pub.subjects, pub.ids, c2) + } + end, err := delivery.DecodeForgeGroupEnd(pub.data[0]) + if err != nil || len(end.Members) != 2 || end.Members[0].Merge != c1 || end.Members[1].Merge != c2 { + t.Errorf("the end = %+v %v, want both members in their merge order", end, err) + } + if rev, ok := done[groupsKey("feat/take-in")]; !ok || rev != 7 { + t.Errorf("performed %v: want groups.feat/take-in kept", done) + } + // Again by hand: written once more, its id not the first's. + if !g.end(ctx, "feat/take-in", 10, gate["feat/take-in"], "20261104T2300") || pub.ids[1] == pub.ids[0] { + t.Errorf("the end again by hand: ids %v", pub.ids) + } +} diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go index c8bb3b00..c630e424 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go @@ -68,6 +68,12 @@ type publisher interface { // redelivered, and the log names its subject and why. A message it could not write stays unacknowledged: the server // delivers it again (the server's ack wait is the only clock). func handle(ctx context.Context, m delivered, pub publisher, trees bridge.Trees, files bridge.Files, kept bridge.Kept, logger *log.Logger) { + handleGrouped(ctx, m, pub, trees, files, kept, nil, logger) +} + +// handleGrouped is handle; a merged message of a delivery group's member names its group (g, ADR 0633 §4; nil names +// none), and the member's merged message written may end the group. +func handleGrouped(ctx context.Context, m delivered, pub publisher, trees bridge.Trees, files bridge.Files, kept bridge.Kept, g *grouper, logger *log.Logger) { msg, err := bridge.ConvertKept(m.Subject(), m.Data(), trees, files, kept) switch { case errors.Is(err, bridge.ErrRetry): @@ -88,10 +94,18 @@ func handle(ctx context.Context, m delivered, pub publisher, trees bridge.Trees, if id := m.Headers().Get("x-event-id"); id != "" { opts = append(opts, jetstream.WithMsgID(id)) } + var member delivery.Forge + grouped := false + if g != nil { + msg.Data, member, grouped = g.member(ctx, msg.Data) + } if _, err := pub.Publish(ctx, forgeSubject(msg.Event), msg.Data, opts...); err != nil { logger.Printf("%s: %s could not be written on FORGE, left for redelivery: %v", m.Subject(), msg.Event, err) return } + if grouped { + g.published(ctx, member) + } if msg.Said != "" { logger.Printf("%s: %s", m.Subject(), msg.Said) } @@ -136,17 +150,25 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err } perf := &performer{call: call, pub: rec, done: kvRevisions{performedKV}, logger: logger} + // A delivery group's member names its group, and its end is said once (ADR 0633 §4), from the gate's merges. + groups := &grouper{send: sendVia(js), done: kvRevisions{performedKV}, logger: logger} watching := make(chan error, 3) for _, w := range []struct { bucket string react func(ctx context.Context, key string, rev uint64, value []byte) - }{{delivery.OwedBucket, perf.owed}, {delivery.MergesBucket, perf.merges}} { + }{{delivery.OwedBucket, perf.owed}, {delivery.MergesBucket, func(ctx context.Context, key string, rev uint64, value []byte) { + perf.merges(ctx, key, rev, value) + groups.ended(ctx, key, rev, value) + }}} { // The gate's buckets come with the gate and its grant; the bridge does not wait for them (no timer), and is started // again by its delivery when the gate arrives. A bucket refused or not answered is the same (issue 1945). kv := process.OptionalBucket(ctx, js, w.bucket, "nothing of it is performed until the bridge starts again", logger.Printf) if kv == nil { continue } + if w.bucket == delivery.MergesBucket { + groups.merges = kvMerges(kv) + } go func() { watching <- watch(ctx, kv, func(key string, rev uint64, value []byte) { hctx, cancel := context.WithTimeout(ctx, syncDeadline) @@ -171,7 +193,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // Bounded: the turn is held, and no sync or other event goes on without it. hctx, cancel := context.WithTimeout(ctx, syncDeadline) defer cancel() - handle(hctx, m, rec, treesVia(hctx, call), filesVia(hctx, call), inputs, logger) + handleGrouped(hctx, m, rec, treesVia(hctx, call), filesVia(hctx, call), inputs, groups, logger) }) }, logger.Printf) if err != nil { diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go index 85df5828..6234a90d 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go @@ -107,11 +107,10 @@ func TestUnreadableEventIsTerminatedAndLogged(t *testing.T) { // 1235/AC5 if !strings.Contains(logs.String(), "mesh.mod.gitea.event.pull.updated") || !strings.Contains(logs.String(), "head") { t.Errorf("log %q names neither the subject nor why", logs.String()) } -} - -func TestUnwrittenMessageIsNotAcked(t *testing.T) { // 1235 (a failed write is redelivered by the server, not retried here) - var logs bytes.Buffer - m, pub := &fakeMsg{subject: "mesh.mod.gitea.event.pull.updated", data: body(nil)}, &fakePub{err: errors.New("no stream")} + // 1235: an unwritten message is not acked (a failed write is redelivered by the server, not retried here); merged here + // from a test of its own, for the test budget. + logs.Reset() + m, pub = &fakeMsg{subject: "mesh.mod.gitea.event.pull.updated", data: body(nil)}, &fakePub{err: errors.New("no stream")} handle(context.Background(), m, pub, nil, someFiles, nil, log.New(&logs, "", 0)) if m.acked != 0 || m.termd != 0 { t.Errorf("acked %d terminated %d, want neither", m.acked, m.termd)