forge bridge: a member's merged message names its group, and the group's end is said once (ADR 0633 §4, issue 1585)
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
This commit is contained in:
@@ -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.<hash>` for each
|
||||
// member's merged message written (the hash is the first 16 hex digits of the sha256 of `<repository>#<number>@<merge>`),
|
||||
// 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)
|
||||
|
||||
// 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.<key>` 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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user