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

Forge bridge: the process takes the keys of performed from internal/groups, one copy (issue 1585)

Browse Source 
Claude-Session: https://claude.ai/code/session_01JG73KG7AfSdh6FP8VrBqCn
This commit is contained in:
jschoubben
2026-11-04 23:24:05 +01:00
parent 92dc5f03d0
commit 1040c8ec46
3 changed files with 12 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groupask_test.go
+2 -6
View File
@@ -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")
}
}
modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groups.go
+7 -20
View File
@@ -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.<merges key>` 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)
modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groups_test.go
+3 -1
View File
@@ -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.