From 1040c8ec46152e808cd30a19b0a268380a05728d Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 23:24:05 +0100 Subject: [PATCH] Forge bridge: the process takes the keys of performed from internal/groups, one copy (issue 1585) Claude-Session: https://claude.ai/code/session_01JG73KG7AfSdh6FP8VrBqCn --- .../groupask_test.go | 8 ++---- .../cmd/mesh-delivery-forge-bridge/groups.go | 27 +++++-------------- .../mesh-delivery-forge-bridge/groups_test.go | 4 ++- 3 files changed, 12 insertions(+), 27 deletions(-) diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groupask_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groupask_test.go index 3d0bb3af..de6ded88 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groupask_test.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groupask_test.go @@ -46,7 +46,7 @@ func TestAnEndAskSaysTheGroupsEndAgainOrWhyNot(t *testing.T) { } { pub, done := &idPub{}, memRevisions{} if tc.published { - done[publishedKey("novox/mesh-sdk", 37, c1)], done[publishedKey("novox/mesh-delivery", 9, c2)] = 1, 1 + done[groups.PublishedKey("novox/mesh-sdk", 37, c1)], done[groups.PublishedKey("novox/mesh-delivery", 9, c2)] = 1, 1 } g := &grouper{send: pub.send, done: done, logger: log.New(&bytes.Buffer{}, "", 0), merges: func(_ context.Context, key string) ([]byte, uint64, bool, error) { @@ -57,12 +57,8 @@ func TestAnEndAskSaysTheGroupsEndAgainOrWhyNot(t *testing.T) { if got.Written != tc.written || !strings.Contains(got.Answer, tc.answer) || !got.Done() || len(pub.ids) != map[bool]int{true: 1}[tc.written] { t.Errorf("%s: %+v, ids %v", tc.name, got, pub.ids) } - if tc.written && pub.ids[0] != groupEndID("feat/take-in", c2)+"."+groups.AgainStamp(ask.At) { + if tc.written && pub.ids[0] != groups.EndID("feat/take-in", c2, groups.AgainStamp(ask.At)) { t.Errorf("%s: the id %s does not carry the ask's time", tc.name, pub.ids[0]) } } - if groupsKey("feat/take-in") != groups.Key("feat/take-in") || publishedKey("novox/mesh-sdk", 37, c1) != groups.PublishedKey("novox/mesh-sdk", 37, c1) || - groupEndID("g", c1) != groups.EndID("g", c1, "") { - t.Error("the tools bundle's keys of performed are not the process's") - } } 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 index 486990b9..824a6a7b 100644 --- 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 @@ -2,10 +2,7 @@ package main import ( "context" - "crypto/sha256" - "encoding/hex" "errors" - "fmt" "log" "slices" "strconv" @@ -14,6 +11,8 @@ import ( "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/delivery" + + groupkeys "mesh-delivery-forge-bridge/internal/groups" ) // A delivery group's merges reach the indexer together (novox/hq ADR 0633 ยง4): the bridge sets the group on each @@ -25,18 +24,6 @@ import ( // 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) @@ -143,7 +130,7 @@ func (g *grouper) member(ctx context.Context, data []byte) ([]byte, delivery.For // 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 { + if err := g.done.Put(ctx, groupkeys.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 } @@ -172,7 +159,7 @@ func (g *grouper) end(ctx context.Context, key string, rev uint64, value []byte, return false } if again == "" { - if _, seen, err := g.done.Get(ctx, groupsKey(key)); err == nil && seen { + if _, seen, err := g.done.Get(ctx, groupkeys.Key(key)); err == nil && seen { return false } } @@ -184,7 +171,7 @@ func (g *grouper) end(ctx context.Context, key string, rev uint64, value []byte, 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 { + if _, ok, err := g.done.Get(ctx, groupkeys.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}) @@ -199,7 +186,7 @@ func (g *grouper) end(ctx context.Context, key string, rev uint64, value []byte, 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) + id := groupkeys.EndID(m.Group, end.Members[len(end.Members)-1].Merge, "") if again != "" { id += "." + again } @@ -207,7 +194,7 @@ func (g *grouper) end(ctx context.Context, key string, rev uint64, value []byte, 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 { + if err := g.done.Put(ctx, groupkeys.Key(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) 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 index 419c259e..0f792c32 100644 --- 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 @@ -8,6 +8,8 @@ import ( "testing" "git.novox.be/novox/mesh-sdk/go/delivery" + + groupkeys "mesh-delivery-forge-bridge/internal/groups" ) // idPub records the subjects, bodies and message ids it is handed. @@ -84,7 +86,7 @@ func TestAGroupsMembersNameItAndItsEndIsSaidOnce(t *testing.T) { // 1585/T16, T1 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 { + if rev, ok := done[groupkeys.Key("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.