From 677f94e9012bac307dc3a84e3091932f71ebcdb5 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 19 Oct 2026 21:56:42 +0200 Subject: [PATCH 1/2] Say which pull requests are open and drop every other candidate, so the candidates hold only open pull requests (issue 1279, ADR 0443) --- .../cmd/mesh-delivery-forge-bridge/held.go | 52 ++++++ .../integration_test.go | 151 ++++++++++++++++-- .../cmd/mesh-delivery-forge-bridge/main.go | 12 +- .../cmd/mesh-delivery-forge-bridge/open.go | 39 ++++- .../mesh-delivery-forge-bridge/open_test.go | 113 ++++++++++++- modules/mesh-delivery-forge-bridge/go.mod | 2 +- modules/mesh-delivery-forge-bridge/go.sum | 4 +- .../internal/bridge/bridge.go | 62 +++++++ .../internal/bridge/bridge_test.go | 55 +++++++ .../cmd/mesh-delivery-indexer/main.go | 9 ++ .../cmd/mesh-delivery-indexer/main_test.go | 7 + modules/mesh-delivery-indexer/go.mod | 2 +- modules/mesh-delivery-indexer/go.sum | 4 +- .../internal/index/candidates.go | 55 +++++++ .../internal/index/index_test.go | 65 ++++++++ 15 files changed, 603 insertions(+), 29 deletions(-) create mode 100644 modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/held.go diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/held.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/held.go new file mode 100644 index 0000000..d51670c --- /dev/null +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/held.go @@ -0,0 +1,52 @@ +package main + +import ( + "context" + "errors" + "fmt" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" + + "mesh-delivery-forge-bridge/internal/bridge" +) + +// forgeStream is the bridge's own stream as the bus names it. +const forgeStream = "FORGE" + +// heldOnForge reads the head FORGE last holds for each pull request: every pull and merged message the stream keeps, in +// order, through an ordered consumer of no name that goes when this returns. It reads up to what is on the stream now. +func heldOnForge(ctx context.Context, js jetstream.JetStream) (bridge.Held, error) { + stream, err := js.Stream(ctx, forgeStream) + if err != nil { + return nil, fmt.Errorf("the stream %s: %w", forgeStream, err) + } + cons, err := stream.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{ + FilterSubjects: []string{forgeSubject("forge.*.*.pull"), forgeSubject("forge.*.*.merged")}, + DeliverPolicy: jetstream.DeliverAllPolicy, + }) + if err != nil { + return nil, fmt.Errorf("reading %s: %w", forgeStream, err) + } + held := bridge.Held{} + for { + batch, err := cons.FetchNoWait(256) + if err != nil { + return nil, fmt.Errorf("reading %s: %w", forgeStream, err) + } + n := 0 + for m := range batch.Messages() { + n++ + if f, err := delivery.DecodeForge(m.Data()); err == nil { + held.Hear(f) + } + } + if err := batch.Error(); err != nil && !errors.Is(err, jetstream.ErrNoMessages) { + return nil, fmt.Errorf("reading %s: %w", forgeStream, err) + } + if n == 0 { + return held, nil + } + } +} diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go index 0a595c7..8fe41ca 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go @@ -10,6 +10,7 @@ import ( "os" "path/filepath" "strings" + "sync" "testing" "time" @@ -68,6 +69,24 @@ func startBus(t *testing.T) *server.Server { // consumer _ on EVENTS, as the controller makes them (mesh-control internal/broker streams.go:192, // nats.go:1387-1396, modstreams.go). It returns a control connection, its JetStream and a context. func startBridge(t *testing.T, openReply string) (*nats.Conn, jetstream.JetStream, context.Context, <-chan error) { + b := startBed(t, openReply, nil) + return b.ctl, b.js, b.ctx, b.done +} + +// bed is a started bridge with the bus it runs against. +type bed struct { + srv *server.Server + ctl *nats.Conn + js jetstream.JetStream + ctx context.Context + done <-chan error + reply func(string) +} + +// startBed is startBridge with a step between the streams and the consumer: prepare writes what the streams hold before +// the bridge's consumer is made, as the controller makes it, from now (ADR 0443 §1), and reply sets the answer of the +// forge's gitea_open_pulls from then on. +func startBed(t *testing.T, openReply string, prepare func(ctx context.Context, js jetstream.JetStream)) *bed { srv := startBus(t) ctl, err := nats.Connect(srv.ClientURL(), nats.UserInfo("ctl", "pw")) if err != nil { @@ -86,16 +105,25 @@ func startBridge(t *testing.T, openReply string) (*nats.Conn, jetstream.JetStrea if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "FORGE", Subjects: []string{"mesh.stream.mesh-delivery-forge-bridge.event.forge.>"}}); err != nil { t.Fatal(err) } + if prepare != nil { + prepare(ctx, js) + } if _, err := js.CreateOrUpdateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{ - Durable: "n1_mesh-delivery-forge-bridge", AckPolicy: jetstream.AckExplicitPolicy, + Durable: "n1_mesh-delivery-forge-bridge", AckPolicy: jetstream.AckExplicitPolicy, DeliverPolicy: jetstream.DeliverNewPolicy, FilterSubjects: []string{"mesh.mod.gitea.event.pull.updated", "mesh.mod.gitea.event.pull.merged", "mesh.mod.gitea.event.pull.closed"}, }); err != nil { t.Fatal(err) } - // A stand-in answers the forge's gitea_open_pulls, which the bridge asks at its start, on its tool subject - // (mesh-control internal/broker/nats.go invokedSubjects) in the reply envelope of mesh-sdk go/process/call.go. - open, err := ctl.Subscribe("mesh.mod.gitea.tool.gitea_open_pulls", func(m *nats.Msg) { _ = m.Respond([]byte(openReply)) }) + // A stand-in answers the forge's gitea_open_pulls, which the bridge asks at its start and after a reconnect, on its tool + // subject (mesh-control internal/broker/nats.go invokedSubjects) in the reply envelope of mesh-sdk go/process/call.go. + var mu sync.Mutex + current := openReply + open, err := ctl.Subscribe("mesh.mod.gitea.tool.gitea_open_pulls", func(m *nats.Msg) { + mu.Lock() + defer mu.Unlock() + _ = m.Respond([]byte(current)) + }) if err != nil { t.Fatal(err) } @@ -114,7 +142,7 @@ func startBridge(t *testing.T, openReply string) (*nats.Conn, jetstream.JetStrea done := make(chan error, 1) go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }() t.Cleanup(stop) - return ctl, js, ctx, done + return &bed{srv: srv, ctl: ctl, js: js, ctx: ctx, done: done, reply: func(s string) { mu.Lock(); current = s; mu.Unlock() }} } // TestBridgeWritesPullMessages starts the program the way the node does (its entry, run, with MESH_BROKER_FILE). @@ -251,7 +279,8 @@ func TestBridgeWritesMergedMessage(t *testing.T) { // 1236/AC4 t.Errorf("the event is not acknowledged: %+v", ci) } -// TestBridgeSaysTheOpenPullsAtStart: the forge holds two open pull requests when the bridge starts; one message each is on FORGE. +// TestBridgeSaysTheOpenPullsAtStart: the forge holds two open pull requests when the bridge starts; one message each is on +// FORGE, and the open message naming both after them (ADR 0443 §5). func TestBridgeSaysTheOpenPullsAtStart(t *testing.T) { // 1237/AC1 // The entries are gitea_open_pulls's (mesh-catalog modules/gitea/client.ts:369-381, head the branch). _, js, ctx, _ := startBridge(t, `{"result":{"pulls":[ @@ -263,12 +292,12 @@ func TestBridgeSaysTheOpenPullsAtStart(t *testing.T) { // 1237/AC1 } var info *jetstream.StreamInfo for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { - if info, err = forge.Info(ctx); err == nil && info.State.Msgs >= 2 { + if info, err = forge.Info(ctx); err == nil && info.State.Msgs >= 3 { break } } - if info.State.Msgs != 2 { - t.Fatalf("FORGE holds %d messages within 5 s, want 2", info.State.Msgs) + if info.State.Msgs != 3 { + t.Fatalf("FORGE holds %d messages within 5 s, want 3", info.State.Msgs) } for _, repo := range []string{"mesh-delivery", "mesh-catalog"} { raw, err := forge.GetLastMsgForSubject(ctx, "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox."+repo+".pull") @@ -293,3 +322,107 @@ func TestBridgeExitsWhenTheOpenPullsCannotBeRead(t *testing.T) { // 1237/AC3 t.Fatal("the bridge did not exit") } } + +// TestBridgeReadsTheForgeAtStartAndAfterAReconnectAndHearsNoOlderEvent starts the bridge whole against a fake forge with two +// open pull requests, one of them with its head already on FORGE, and a bus whose EVENTS holds older pull events from +// before its consumer was made. At start it writes the one pull message that differs and the open message naming both; +// the bus connection closed, it reads the forge again and writes the open message alone, since every head is held now; +// for the older events it writes nothing (ADR 0443 §1, §5). +func TestBridgeReadsTheForgeAtStartAndAfterAReconnectAndHearsNoOlderEvent(t *testing.T) { // 1279/AC8 + h2 := strings.Repeat("a", 40) + pullSubject := func(repo string) string { + return "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox." + repo + ".pull" + } + openSubject := "mesh.stream.mesh-delivery-forge-bridge.event.forge.open" + b := startBed(t, `{"result":{"pulls":[ + {"owner":"novox","repo":"mesh-delivery","number":12,"base":"main","head":"feat/x","head_sha":"`+head+`","html_url":"h","required":[],"statuses":[],"checks":{},"labels":[]}, + {"owner":"novox","repo":"mesh-catalog","number":300,"base":"main","head":"feat/x","head_sha":"`+h2+`","html_url":"h","required":[],"statuses":[],"checks":{},"labels":[]}]},"node":"n0"}`, + func(ctx context.Context, js jetstream.JetStream) { + // Older events: a pull request long closed and one long updated, from before the bridge's consumer. + for subject, ev := range map[string]map[string]any{ + "mesh.mod.gitea.event.pull.updated": {"owner": "novox", "repo": "mesh-old", "number": 5, "head": "feat/old", "head_sha": h2}, + "mesh.mod.gitea.event.pull.closed": {"owner": "novox", "repo": "mesh-older", "number": 6, "head": "feat/older", "head_sha": h2}, + } { + raw, _ := json.Marshal(ev) + if _, err := js.Publish(ctx, subject, raw); err != nil { + t.Fatal(err) + } + } + // What FORGE already holds: the head of mesh-delivery#12. + raw, _ := delivery.EncodeForge(delivery.Forge{Kind: delivery.ForgePull, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: head, Action: delivery.PullUpdated}) + if _, err := js.Publish(ctx, pullSubject("mesh-delivery"), raw); err != nil { + t.Fatal(err) + } + }) + forge, err := b.js.Stream(b.ctx, "FORGE") + if err != nil { + t.Fatal(err) + } + waitMsgs := func(want uint64) { + t.Helper() + var info *jetstream.StreamInfo + for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + if info, err = forge.Info(b.ctx); err == nil && info.State.Msgs >= want { + break + } + } + time.Sleep(300 * time.Millisecond) // and nothing more arrives + if info, err = forge.Info(b.ctx); err != nil || info.State.Msgs != want { + t.Fatalf("FORGE holds %d messages (%v), want %d", info.State.Msgs, err, want) + } + } + named := func(seq uint64) []delivery.OpenPull { + t.Helper() + raw, err := forge.GetMsg(b.ctx, seq) + if err != nil || raw.Subject != openSubject { + t.Fatalf("message %d: %v %v, want the open message", seq, raw, err) + } + f, err := delivery.DecodeForge(raw.Data) + if err != nil { + t.Fatal(err) + } + return f.Open + } + bothOpen := []delivery.OpenPull{{Repository: "novox/mesh-catalog", Number: 300}, {Repository: "novox/mesh-delivery", Number: 12}} + same := func(got []delivery.OpenPull) bool { + return len(got) == 2 && got[0] == bothOpen[0] && got[1] == bothOpen[1] + } + + // At start: the held head (message 1), the one that differs (2), the open message naming both (3). + waitMsgs(3) + if raw, err := forge.GetMsg(b.ctx, 2); err != nil || raw.Subject != pullSubject("mesh-catalog") { + t.Fatalf("message 2 = %v %v, want mesh-catalog's pull message", raw, err) + } + if got := named(3); !same(got) { + t.Fatalf("the open message names %v, want %v", got, bothOpen) + } + + // The connection closes and reopens: the forge is read again, every head is held, the open message alone follows. + conns, err := b.srv.Connz(&server.ConnzOptions{}) + if err != nil { + t.Fatal(err) + } + closed := 0 + for _, c := range conns.Conns { + if c.Name == "n1.mesh-delivery-forge-bridge" { + if err := b.srv.DisconnectClientByID(c.Cid); err != nil { + t.Fatal(err) + } + closed++ + } + } + if closed != 1 { + t.Fatalf("%d connections of the bridge found", closed) + } + waitMsgs(4) + if got := named(4); !same(got) { + t.Fatalf("after the reconnect the open message names %v, want %v", got, bothOpen) + } + + // For the older events nothing was written. + for _, repo := range []string{"mesh-old", "mesh-older"} { + if raw, err := forge.GetLastMsgForSubject(b.ctx, pullSubject(repo)); err == nil { + t.Errorf("a message for %s's older event: %+v", repo, raw) + } + } +} 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 68e70ac..e76b126 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 @@ -18,6 +18,7 @@ import ( "os" "os/signal" "strings" + "sync" "syscall" "time" @@ -185,16 +186,19 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } // What the forge holds open, said before any event is heard, and again at each reconnect (ADR 0429 §13). call := callerOn(conn, cred.User) - sync := func(ctx context.Context) error { return syncOpen(ctx, call, js, logger) } - if err := sync(ctx); err != nil { + // No forge event is heard while the open pull requests are read and said (ADR 0443 §5): events and syncs take turns. + var turn sync.Mutex + held := func(ctx context.Context) (bridge.Held, error) { return heldOnForge(ctx, js) } + said := func(ctx context.Context) error { return syncOpen(ctx, call, js, held, &turn, logger) } + if err := said(ctx); err != nil { return err } - go onReconnect(ctx, again, sync, logger) + go onReconnect(ctx, again, said, logger) cons, err := js.Consumer(ctx, eventsStream, cred.Node+"_"+cred.Module) if err != nil { return fmt.Errorf("the consumer %s_%s on %s: %w", cred.Node, cred.Module, eventsStream, err) } - cc, err := cons.Consume(func(m jetstream.Msg) { handle(ctx, m, js, treesVia(ctx, call), logger) }, + cc, err := cons.Consume(func(m jetstream.Msg) { hear(&turn, func() { handle(ctx, m, js, treesVia(ctx, call), logger) }) }, jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) })) if err != nil { return err diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/open.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/open.go index 0d13d24..8a9c576 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/open.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/open.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "log" + "sync" "mesh-delivery-forge-bridge/internal/bridge" ) @@ -12,10 +13,27 @@ import ( // gitea_open_pulls: no arguments, answers {pulls: [...]}). const openPullsTool = "gitea.gitea_open_pulls" -// syncOpen reads the open pull requests from the forge and writes one pull message for each: what makes the indexer's -// candidates match the forge after a restart or a gap in the forge's events (ADR 0429 §13). An error names the call and -// the forge's words. -func syncOpen(ctx context.Context, call caller, pub publisher, logger *log.Logger) error { +// heldReader reads the head FORGE last holds for each pull request. +type heldReader func(ctx context.Context) (bridge.Held, error) + +// hear runs one reaction to a forge event while no sync of the open pull requests is between reading the forge and +// writing the open message (novox/hq ADR 0443 §5): an event heard meanwhile waits, and is written after it, so a pull +// request opened after the read is added again by its own message and never dropped by the open message. +func hear(mu *sync.Mutex, react func()) { + mu.Lock() + defer mu.Unlock() + react() +} + +// syncOpen reads the open pull requests from the forge and writes a pull message for each whose head FORGE does not +// hold, then one open message naming every one it read, those whose head did not differ among them: what makes the +// indexer's candidates match the forge after a restart or a gap in the forge's events (ADR 0429 §13, ADR 0443 §5). It +// holds mu from before it reads the forge until the open message is written, so no event is heard between. What FORGE +// holds that cannot be read is logged, and every open pull request is written. An error names the call and the forge's +// words. +func syncOpen(ctx context.Context, call caller, pub publisher, held heldReader, mu *sync.Mutex, logger *log.Logger) error { + mu.Lock() + defer mu.Unlock() raw, err := call(ctx, openPullsTool, struct{}{}) if err != nil { return fmt.Errorf("%s: %w", openPullsTool, err) @@ -24,15 +42,24 @@ func syncOpen(ctx context.Context, call caller, pub publisher, logger *log.Logge if err != nil { return fmt.Errorf("%s: %w", openPullsTool, err) } + open, err := bridge.OpenMessage(raw) + if err != nil { + return fmt.Errorf("%s: %w", openPullsTool, err) + } for _, s := range skipped { logger.Printf("an open pull request is left out: %s", s) } - for _, m := range msgs { + if h, err := held(ctx); err != nil { + logger.Printf("what FORGE holds could not be read, so every open pull request is written: %v", err) + } else { + msgs = h.Differing(msgs) + } + for _, m := range append(msgs, open) { if _, err := pub.Publish(ctx, forgeSubject(m.Event), m.Data); err != nil { return fmt.Errorf("the open pull request message %s could not be written on FORGE: %w", m.Event, err) } } - logger.Printf("%d open pull requests said on FORGE", len(msgs)) + logger.Printf("%d pull messages and the open message written on FORGE", len(msgs)) return nil } diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/open_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/open_test.go index 33fb81d..62d5b70 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/open_test.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/open_test.go @@ -11,7 +11,11 @@ import ( "testing" "time" + "github.com/nats-io/nats.go/jetstream" + "git.novox.be/novox/mesh-sdk/go/delivery" + + "mesh-delivery-forge-bridge/internal/bridge" ) // openAnswer is gitea_open_pulls's answer as mesh-catalog modules/gitea/client.ts:369-381 builds its entries inside {pulls}. @@ -30,14 +34,14 @@ func TestSyncOpenWritesOneUpdatedMessagePerOpenPull(t *testing.T) { // 1237/AC1, keys = append(keys, key) return openAnswer(), nil } - if err := syncOpen(context.Background(), call, pub, log.New(&logs, "", 0)); err != nil { + if err := syncOpen(context.Background(), call, pub, nothingHeld, &sync.Mutex{}, log.New(&logs, "", 0)); err != nil { t.Fatal(err) } if len(keys) != 1 || keys[0] != "gitea.gitea_open_pulls" { t.Errorf("calls %v", keys) } - if len(pub.subjects) != 2 { - t.Fatalf("%d messages written (%v), want two: the pull request into feat/y gets none", len(pub.subjects), pub.subjects) + if len(pub.subjects) != 3 { + t.Fatalf("%d messages written (%v), want two pull messages and the open message: the pull request into feat/y gets none", len(pub.subjects), pub.subjects) } want := []string{"mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-delivery.pull", "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-catalog.pull"} for i, w := range want { @@ -48,12 +52,113 @@ func TestSyncOpenWritesOneUpdatedMessagePerOpenPull(t *testing.T) { // 1237/AC1, } } +func nothingHeld(context.Context) (bridge.Held, error) { return bridge.Held{}, nil } + +// 1279/AC6: the sync writes a pull message only for a head that differs from what FORGE holds, then one open message +// naming every open pull request into trunk it read, those whose head did not differ among them. +func TestSyncOpenWritesPullsForNewHeadsThenOneOpenMessageNamingAll(t *testing.T) { // 1279/AC6 + pub := &fakePub{} + call := func(context.Context, string, any) (json.RawMessage, error) { return openAnswer(), nil } + held := func(context.Context) (bridge.Held, error) { + h := bridge.Held{} + h.Hear(delivery.Forge{Kind: delivery.ForgePull, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: head, Action: delivery.PullUpdated}) + return h, nil + } + var logs bytes.Buffer + if err := syncOpen(context.Background(), call, pub, held, &sync.Mutex{}, log.New(&logs, "", 0)); err != nil { + t.Fatal(err) + } + if len(pub.subjects) != 2 { + t.Fatalf("%d messages written (%v), want the one pull message that differs and the open message", len(pub.subjects), pub.subjects) + } + if pub.subjects[0] != "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-catalog.pull" { + t.Errorf("first message on %s, want mesh-catalog's pull message", pub.subjects[0]) + } + if pub.subjects[1] != "mesh.stream.mesh-delivery-forge-bridge.event.forge.open" { + t.Fatalf("last message on %s, want the open message after the pull messages", pub.subjects[1]) + } + f, err := delivery.DecodeForge(pub.data[1]) + want := []delivery.OpenPull{{Repository: "novox/mesh-catalog", Number: 300}, {Repository: "novox/mesh-delivery", Number: 12}} + if err != nil || len(f.Open) != 2 || f.Open[0] != want[0] || f.Open[1] != want[1] { + t.Errorf("open message = %+v err %v, want it to name %v, the pull request whose head did not differ among them", f, err, want) + } +} + +// What FORGE holds cannot be read: every open pull request is written, as before, and the open message follows; the +// indexer's writes are idempotent, so the fault costs messages and no state. +func TestSyncOpenWritesAllWhenFORGECannotBeRead(t *testing.T) { // 1279/AC6 + pub := &fakePub{} + call := func(context.Context, string, any) (json.RawMessage, error) { return openAnswer(), nil } + held := func(context.Context) (bridge.Held, error) { return nil, errors.New("no consumer on FORGE") } + var logs bytes.Buffer + if err := syncOpen(context.Background(), call, pub, held, &sync.Mutex{}, log.New(&logs, "", 0)); err != nil { + t.Fatal(err) + } + if len(pub.subjects) != 3 || !strings.Contains(logs.String(), "no consumer on FORGE") { + t.Errorf("messages %v, log %q", pub.subjects, logs.String()) + } +} + +// 1279/AC6: no forge event is heard between reading the forge and writing the open message. +func TestNoForgeEventIsHeardWhileTheOpenPullsAreSaid(t *testing.T) { // 1279/AC6 + pub := &lockedPub{} + reading := make(chan struct{}) + release := make(chan struct{}) + call := func(context.Context, string, any) (json.RawMessage, error) { + close(reading) + <-release + return openAnswer(), nil + } + var mu sync.Mutex + synced := make(chan error, 1) + go func() { + synced <- syncOpen(context.Background(), call, pub, nothingHeld, &mu, log.New(&bytes.Buffer{}, "", 0)) + }() + <-reading + // An event arrives while the forge is being read. + heard := make(chan struct{}) + go func() { + hear(&mu, func() { _, _ = pub.Publish(context.Background(), "event", nil); close(heard) }) + }() + select { + case <-heard: + t.Fatal("an event was heard while the forge was read") + case <-time.After(100 * time.Millisecond): + } + close(release) + if err := <-synced; err != nil { + t.Fatal(err) + } + <-heard + got := pub.snapshot() + if len(got) != 4 || got[2] != "mesh.stream.mesh-delivery-forge-bridge.event.forge.open" || got[3] != "event" { + t.Errorf("order of writes: %v, want the event after the open message", got) + } +} + +type lockedPub struct { + mu sync.Mutex + subjects []string +} + +func (p *lockedPub) Publish(_ context.Context, subject string, _ []byte, _ ...jetstream.PublishOpt) (*jetstream.PubAck, error) { + p.mu.Lock() + defer p.mu.Unlock() + p.subjects = append(p.subjects, subject) + return &jetstream.PubAck{}, nil +} +func (p *lockedPub) snapshot() []string { + p.mu.Lock() + defer p.mu.Unlock() + return append([]string(nil), p.subjects...) +} + func TestSyncOpenFailureNamesTheCallAndTheForgesWords(t *testing.T) { // 1237/AC3 pub := &fakePub{} call := func(context.Context, string, any) (json.RawMessage, error) { return nil, errors.New("the forge said 502") } - err := syncOpen(context.Background(), call, pub, log.New(&bytes.Buffer{}, "", 0)) + err := syncOpen(context.Background(), call, pub, nothingHeld, &sync.Mutex{}, log.New(&bytes.Buffer{}, "", 0)) if err == nil || !strings.Contains(err.Error(), "gitea.gitea_open_pulls") || !strings.Contains(err.Error(), "the forge said 502") { t.Errorf("err = %v", err) } diff --git a/modules/mesh-delivery-forge-bridge/go.mod b/modules/mesh-delivery-forge-bridge/go.mod index 971f980..fb59585 100644 --- a/modules/mesh-delivery-forge-bridge/go.mod +++ b/modules/mesh-delivery-forge-bridge/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-forge-bridge go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019124204-9ade199740e3 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019195003-4580d80a32d9 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-forge-bridge/go.sum b/modules/mesh-delivery-forge-bridge/go.sum index cb9936c..f74bae2 100644 --- a/modules/mesh-delivery-forge-bridge/go.sum +++ b/modules/mesh-delivery-forge-bridge/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019124204-9ade199740e3 h1:Mu2QiGbL/sZFU5ya4lGUyQOB5OKXPy/KB40XfXTf2EU= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019124204-9ade199740e3/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019195003-4580d80a32d9 h1:XL1dbtSleGAixg6FX4gEnSG/AJxRG/tWw6/7U1RuzHc= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019195003-4580d80a32d9/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go index 4b450de..7a3ebed 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go @@ -6,6 +6,8 @@ import ( "errors" "fmt" "regexp" + "slices" + "strings" "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -186,3 +188,63 @@ func OpenPulls(answer []byte) (msgs []Message, skipped []string, err error) { } return msgs, skipped, nil } + +// Held is the head FORGE last holds for each pull request, by `/#`: what the bridge compares an +// open pull request's head with, so it writes a pull message only for a head that differs (novox/hq ADR 0443 §5). +type Held map[string]string + +func heldKey(repository string, number int) string { return fmt.Sprintf("%s#%d", repository, number) } + +// Hear takes one message of FORGE into what is held: a pull message holds its head, a closed one or a merged message +// holds nothing for its pull request. Other kinds change nothing. +func (h Held) Hear(f delivery.Forge) { + key := heldKey(f.Repository, f.Number) + switch { + case f.Kind == delivery.ForgePull && f.Action != delivery.PullClosed: + h[key] = f.Head + case f.Kind == delivery.ForgePull || f.Kind == delivery.ForgeMerged: + delete(h, key) + } +} + +// Differing keeps the pull messages whose head FORGE does not hold, in their order. +func (h Held) Differing(msgs []Message) []Message { + var out []Message + for _, m := range msgs { + if f, err := delivery.DecodeForge(m.Data); err == nil && h[heldKey(f.Repository, f.Number)] == f.Head { + continue + } + out = append(out, m) + } + return out +} + +// OpenMessage turns the answer of gitea_open_pulls into the open message: every open pull request into the trunk, by +// repository and number, whether its head reads or not and whether FORGE holds it or not, in repository and number +// order. A pull request into another branch is not named. An answer that names none is a message too: nothing is open. +func OpenMessage(answer []byte) (Message, error) { + var a struct { + Pulls *[]openPull `json:"pulls"` + } + if err := json.Unmarshal(answer, &a); err != nil || a.Pulls == nil { + return Message{}, fmt.Errorf("the answer of gitea_open_pulls is not {pulls: [...]}: %v", err) + } + open := []delivery.OpenPull{} + for _, p := range *a.Pulls { + if p.Base != Trunk || p.Owner == "" || p.Repo == "" || p.Number <= 0 { + continue + } + open = append(open, delivery.OpenPull{Repository: p.Owner + "/" + p.Repo, Number: p.Number}) + } + slices.SortFunc(open, func(a, b delivery.OpenPull) int { + if c := strings.Compare(a.Repository, b.Repository); c != 0 { + return c + } + return a.Number - b.Number + }) + body, err := delivery.EncodeForge(delivery.Forge{Kind: delivery.ForgeOpen, Open: open}) + if err != nil { + return Message{}, fmt.Errorf("the open pull requests cannot be said: %w", err) + } + return Message{Event: delivery.OpenEvent(), Data: body}, nil +} diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge_test.go b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge_test.go index 836ba0b..507ad48 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge_test.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge_test.go @@ -323,3 +323,58 @@ func TestOpenPullsUnreadableEntryIsNamedAndTheRestGo(t *testing.T) { // 1237 } } } + +func TestOpenMessageNamesEveryOpenPullIntoTheTrunk(t *testing.T) { // 1279/AC6 + h2 := strings.Repeat("a", 40) + answer := openAnswer(openEntry("mesh-delivery", 12, "main", "feat/x", head), openEntry("mesh-catalog", 300, "main", "feat/x", h2), + openEntry("mesh-catalog", 301, "feat/y", "feat/z", h2), openEntry("a", 1, "main", "b", "short")) + m, err := OpenMessage(answer) + if err != nil || m.Event != "forge.open" { + t.Fatalf("message %+v err %v", m, err) + } + f, err := delivery.DecodeForge(m.Data) + // Every one into the trunk, whether its head reads or not (it is open); none into another branch; in a fixed order. + want := []delivery.OpenPull{{Repository: "novox/a", Number: 1}, {Repository: "novox/mesh-catalog", Number: 300}, {Repository: "novox/mesh-delivery", Number: 12}} + if err != nil || f.Kind != delivery.ForgeOpen || len(f.Open) != 3 || f.Open[0] != want[0] || f.Open[1] != want[1] || f.Open[2] != want[2] { + t.Errorf("open message = %+v err %v, want %v", f, err, want) + } + // Nothing open is a message too: the indexer then drops every candidate. + m, err = OpenMessage([]byte(`{"pulls":[]}`)) + if f, derr := delivery.DecodeForge(m.Data); err != nil || derr != nil || f.Kind != delivery.ForgeOpen || len(f.Open) != 0 { + t.Errorf("empty answer: %+v %v %v", f, err, derr) + } + for _, bad := range []string{`{"pulls":null}`, `{}`, `x`} { + if _, err := OpenMessage([]byte(bad)); err == nil { + t.Errorf("%s was taken as an answer", bad) + } + } +} + +func TestOnlyAHeadFORGEDoesNotHoldIsWritten(t *testing.T) { // 1279/AC6 + h2 := strings.Repeat("a", 40) + msgs, _, err := OpenPulls(openAnswer(openEntry("mesh-delivery", 12, "main", "feat/x", head), openEntry("mesh-catalog", 300, "main", "feat/x", h2))) + if err != nil { + t.Fatal(err) + } + held := Held{} + held.Hear(delivery.Forge{Kind: delivery.ForgePull, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: head, Action: delivery.PullUpdated}) + got := held.Differing(msgs) + if len(got) != 1 || got[0].Event != "forge.novox.mesh-catalog.pull" { + t.Fatalf("written: %+v, want only mesh-catalog#300", got) + } + // Another head for the same pull request differs; a closed or merged one is no longer held. + held.Hear(delivery.Forge{Kind: delivery.ForgePull, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: h2, Action: delivery.PullUpdated}) + if got := held.Differing(msgs[:1]); len(got) != 1 { + t.Errorf("a pull request held at another head was left out") + } + held.Hear(delivery.Forge{Kind: delivery.ForgePull, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: head, Action: delivery.PullUpdated}) + held.Hear(delivery.Forge{Kind: delivery.ForgePull, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: head, Action: delivery.PullClosed}) + if got := held.Differing(msgs[:1]); len(got) != 1 { + t.Errorf("a closed pull request is still held") + } + held.Hear(delivery.Forge{Kind: delivery.ForgePull, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: head, Action: delivery.PullOpened}) + held.Hear(delivery.Forge{Kind: delivery.ForgeMerged, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: head, Merge: "m", Root: "r", Trees: map[string]string{"": "t"}}) + if got := held.Differing(msgs[:1]); len(got) != 1 { + t.Errorf("a merged pull request is still held") + } +} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index 245d524..5465157 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -121,6 +121,15 @@ func (s kvStore) Delete(ctx context.Context, key string, revision uint64) error return conflict(s.kv.Delete(ctx, key, jetstream.LastRevision(revision))) } +// Keys is every key of the bucket; a bucket with none is an empty list. +func (s kvStore) Keys(ctx context.Context) ([]string, error) { + keys, err := s.kv.Keys(ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return nil, nil + } + return keys, err +} + // delivered is the part of a JetStream message the indexer uses. type delivered interface { Subject() string diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go index ed5c251..4b8cda5 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go @@ -29,6 +29,13 @@ func (m *memStore) Get(_ context.Context, k string) ([]byte, uint64, bool, error v, ok := m.vals[k] return v, 1, ok, nil } +func (m *memStore) Keys(context.Context) ([]string, error) { + var keys []string + for k := range m.vals { + keys = append(keys, k) + } + return keys, nil +} func (m *memStore) Create(_ context.Context, k string, v []byte) error { m.vals[k] = v; return nil } func (m *memStore) Update(_ context.Context, k string, v []byte, _ uint64) error { m.vals[k] = v diff --git a/modules/mesh-delivery-indexer/go.mod b/modules/mesh-delivery-indexer/go.mod index 42bc36a..d3b0aad 100644 --- a/modules/mesh-delivery-indexer/go.mod +++ b/modules/mesh-delivery-indexer/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-indexer go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019124204-9ade199740e3 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019195003-4580d80a32d9 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-indexer/go.sum b/modules/mesh-delivery-indexer/go.sum index cb9936c..f74bae2 100644 --- a/modules/mesh-delivery-indexer/go.sum +++ b/modules/mesh-delivery-indexer/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019124204-9ade199740e3 h1:Mu2QiGbL/sZFU5ya4lGUyQOB5OKXPy/KB40XfXTf2EU= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019124204-9ade199740e3/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019195003-4580d80a32d9 h1:XL1dbtSleGAixg6FX4gEnSG/AJxRG/tWw6/7U1RuzHc= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019195003-4580d80a32d9/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 6747a61..8bf63af 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -26,6 +26,8 @@ type Store interface { Update(ctx context.Context, key string, value []byte, revision uint64) error // Delete removes the key if its revision is still revision (ErrConflict when not). Delete(ctx context.Context, key string, revision uint64) error + // Keys is every key the store holds now; none is an empty list, not an error. + Keys(ctx context.Context) ([]string, error) } // trunkLike are the branches that are no group: a pull request from one is nobody's delivery group (the rule of @@ -55,6 +57,9 @@ func (ix *Indexer) logf(format string, args ...any) { // by compare-and-set on the revision read, and read again on a conflict. A lone pull request is a group of one. A // merged message also writes the fingerprint of each module of its repository (modules.go). func (ix *Indexer) Apply(ctx context.Context, f delivery.Forge) error { + if f.Kind == delivery.ForgeOpen { + return ix.applyOpen(ctx, f) + } if f.Kind != delivery.ForgePull && f.Kind != delivery.ForgeMerged { return nil } @@ -92,6 +97,56 @@ func (ix *Indexer) Apply(ctx context.Context, f delivery.Forge) error { }) } +// applyOpen takes an open message into the candidates (novox/hq ADR 0443 §5): every pull request it does not name is +// taken out of its group, and a group left with none is deleted. What the message names is left as it is. A pull request +// opened after the bridge read the forge is heard after this message and added again by its own. +func (ix *Indexer) applyOpen(ctx context.Context, f delivery.Forge) error { + named := make(map[delivery.OpenPull]bool, len(f.Open)) + for _, p := range f.Open { + named[p] = true + } + keys, err := ix.Store.Keys(ctx) + if err != nil { + return fmt.Errorf("listing the candidates: %w", err) + } + for _, key := range keys { + err := write(ctx, ix.Store, key, func(value []byte, found bool) ([]byte, bool, error) { + if !found { + return nil, false, nil + } + c, err := delivery.DecodeCandidate(value) + if err != nil { + return nil, false, fmt.Errorf("candidates.%s holds what is not a candidate: %w", key, err) + } + kept := slices.DeleteFunc(slices.Clone(c.Pulls), func(p delivery.CandidatePull) bool { + return !named[delivery.OpenPull{Repository: p.Repository, Number: p.Number}] + }) + switch { + case len(kept) == len(c.Pulls): + return nil, false, nil + case len(kept) == 0: + ix.logf("%s: no pull request of the group is open, so no candidate", key) + return nil, true, nil + } + heads := make([]delivery.Head, len(kept)) + for i, p := range kept { + heads[i] = p.Head + } + hash, err := delivery.HeadsHash(heads) + if err != nil { + return nil, false, fmt.Errorf("%w: candidates.%s: %v", ErrUnreadable, key, err) + } + c.Pulls, c.HeadsHash = kept, hash + data, err := delivery.EncodeCandidate(c) + return data, false, err + }) + if err != nil { + return err + } + } + return nil +} + // write changes one key of a store by compare-and-set: compute is given the value read (found says there is one) and // answers the new value, or del to remove the key, or neither for no change. A conflict reads again, up to writeTries. func write(ctx context.Context, st Store, key string, compute func(value []byte, found bool) (next []byte, del bool, err error)) error { diff --git a/modules/mesh-delivery-indexer/internal/index/index_test.go b/modules/mesh-delivery-indexer/internal/index/index_test.go index 2a01d8a..2493431 100644 --- a/modules/mesh-delivery-indexer/internal/index/index_test.go +++ b/modules/mesh-delivery-indexer/internal/index/index_test.go @@ -32,6 +32,13 @@ func (m *memStore) Get(_ context.Context, key string) ([]byte, uint64, bool, err v, ok := m.vals[key] return v, m.revs[key], ok, nil } +func (m *memStore) Keys(context.Context) ([]string, error) { + var keys []string + for k := range m.vals { + keys = append(keys, k) + } + return keys, nil +} func (m *memStore) hook() { if h := m.beforeWrite; h != nil { m.beforeWrite = nil @@ -306,3 +313,61 @@ func TestAnUnusableBranchNameIsUnreadable(t *testing.T) { t.Errorf("err = %v", err) } } + +func open(pulls ...delivery.OpenPull) delivery.Forge { + return delivery.Forge{V: delivery.Version, Kind: delivery.ForgeOpen, Open: pulls} +} + +// 1279/AC7 (ADR 0443 §5): on an open message, candidates lose every pull request it does not name, a group left empty is +// deleted, and a pull request added by a later message stays. +func TestAnOpenMessageDropsEveryCandidateItDoesNotName(t *testing.T) { // 1279/AC7 + ix, st, _ := newIndexer(t) + ctx := context.Background() + for _, f := range []delivery.Forge{ + pull("novox/mesh-delivery", 12, "feat/x", h1, "modules/a"), pull("novox/mesh-catalog", 300, "feat/x", h2, "modules/c"), + pull("novox/mesh-delivery", 13, "feat/y", h2, "modules/a"), + pull("novox/mesh-catalog", 280, "feat/old", h3, "modules/c"), + } { + if err := ix.Apply(ctx, f); err != nil { + t.Fatal(err) + } + } + before, _ := read(t, st, "feat/x") + + // Open: mesh-delivery#12 only (and a pull request no group holds, which changes nothing). + if err := ix.Apply(ctx, open(delivery.OpenPull{Repository: "novox/mesh-delivery", Number: 12}, delivery.OpenPull{Repository: "novox/mesh-catalog", Number: 999})); err != nil { + t.Fatal(err) + } + c, ok := read(t, st, "feat/x") + hash, _ := delivery.HeadsHash([]delivery.Head{{Repository: "novox/mesh-delivery", Number: 12, Head: h1}}) + if !ok || len(c.Pulls) != 1 || c.Pulls[0].Number != 12 || c.Pulls[0].Head.Head != h1 || strings.Join(c.Pulls[0].Modules, ",") != "a" || c.HeadsHash != hash || c.HeadsHash == before.HeadsHash { + t.Fatalf("feat/x after the open message = %+v, want only mesh-delivery#12 with its modules and the heads hash of it alone", c) + } + for _, gone := range []string{"feat/y", "feat/old"} { + if _, ok := st.vals[gone]; ok { + t.Errorf("the group %s is left with no open pull request and was not deleted", gone) + } + } + + // A pull request opened after the read is heard as a message after the open message and stays. + if err := ix.Apply(ctx, pull("novox/mesh-catalog", 300, "feat/x", h2, "modules/c")); err != nil { + t.Fatal(err) + } + if c, _ := read(t, st, "feat/x"); len(c.Pulls) != 2 { + t.Errorf("a pull request added after the open message is not kept: %+v", c) + } + // The same open message again changes nothing (no write), and an empty one drops every group. + writes := st.writes + if err := ix.Apply(ctx, open(delivery.OpenPull{Repository: "novox/mesh-delivery", Number: 12}, delivery.OpenPull{Repository: "novox/mesh-catalog", Number: 300})); err != nil { + t.Fatal(err) + } + if st.writes != writes { + t.Errorf("an open message that names every candidate wrote %d times", st.writes-writes) + } + if err := ix.Apply(ctx, open()); err != nil { + t.Fatal(err) + } + if len(st.vals) != 0 { + t.Errorf("after an empty open message the candidates hold %v", st.vals) + } +} From ca667af562b0849df9caf5f13dfa2007e4c13fba Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 19 Oct 2026 22:09:41 +0200 Subject: [PATCH 2/2] Keep the head last written per pull request in a bucket of the bridge's own instead of scanning FORGE, and test the bridge under its composed permissions (issue 1279, ADR 0443) --- .../cmd/mesh-delivery-forge-bridge/held.go | 57 ++++------ .../integration_test.go | 89 +++++++++++---- .../cmd/mesh-delivery-forge-bridge/main.go | 19 +++- .../cmd/mesh-delivery-forge-bridge/open.go | 93 ++++++++++++--- .../mesh-delivery-forge-bridge/open_test.go | 107 +++++++++++++++--- .../internal/bridge/bridge.go | 33 ++---- .../internal/bridge/bridge_test.go | 38 +++---- .../manifest_test.go | 14 ++- .../mesh-delivery-forge-bridge/module.json | 3 + 9 files changed, 313 insertions(+), 140 deletions(-) diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/held.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/held.go index d51670c..050935a 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/held.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/held.go @@ -3,50 +3,33 @@ package main import ( "context" "errors" - "fmt" "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/delivery" - - "mesh-delivery-forge-bridge/internal/bridge" ) -// forgeStream is the bridge's own stream as the bus names it. -const forgeStream = "FORGE" +// headsBucket is the module's state bucket heads, `_` as the indexer's buckets are named: the head last +// written on FORGE for each pull request (manifest state entry heads, novox/hq ADR 0443 §5). +const headsBucket = delivery.ForgeBridgeModule + "_heads" -// heldOnForge reads the head FORGE last holds for each pull request: every pull and merged message the stream keeps, in -// order, through an ordered consumer of no name that goes when this returns. It reads up to what is on the stream now. -func heldOnForge(ctx context.Context, js jetstream.JetStream) (bridge.Held, error) { - stream, err := js.Stream(ctx, forgeStream) +// kvHeads is the bucket heads as the bridge's headStore. +type kvHeads struct{ kv jetstream.KeyValue } + +func (s kvHeads) Get(ctx context.Context, key string) (string, bool, error) { + e, err := s.kv.Get(ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return "", false, nil + } if err != nil { - return nil, fmt.Errorf("the stream %s: %w", forgeStream, err) - } - cons, err := stream.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{ - FilterSubjects: []string{forgeSubject("forge.*.*.pull"), forgeSubject("forge.*.*.merged")}, - DeliverPolicy: jetstream.DeliverAllPolicy, - }) - if err != nil { - return nil, fmt.Errorf("reading %s: %w", forgeStream, err) - } - held := bridge.Held{} - for { - batch, err := cons.FetchNoWait(256) - if err != nil { - return nil, fmt.Errorf("reading %s: %w", forgeStream, err) - } - n := 0 - for m := range batch.Messages() { - n++ - if f, err := delivery.DecodeForge(m.Data()); err == nil { - held.Hear(f) - } - } - if err := batch.Error(); err != nil && !errors.Is(err, jetstream.ErrNoMessages) { - return nil, fmt.Errorf("reading %s: %w", forgeStream, err) - } - if n == 0 { - return held, nil - } + return "", false, err } + return string(e.Value()), true, nil } + +func (s kvHeads) Put(ctx context.Context, key, head string) error { + _, err := s.kv.PutString(ctx, key, head) + return err +} + +func (s kvHeads) Delete(ctx context.Context, key string) error { return s.kv.Delete(ctx, key) } diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go index 8fe41ca..6769ff0 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go @@ -22,7 +22,14 @@ import ( ) // busConf is the shape of the composed accounts file (mesh-controller internal/broker): one account with JetStream, -// the controller's user and the bridge's own principal .. +// the controller's user, and the bridge's own principal . with the permissions the controller composes for +// it, and no others, so a permission the bridge lacks fails a test here and not on the mesh (issue 1279). +// +// The lists are what mesh-controller's broker.PermissionsFor answers (internal/broker/nats.go:577) for a module principal +// with the manifest of module.json: consumes the three gitea.pull events, invokes gitea.gitea_open_pulls and +// gitea.gitea_commit_trees, state heads, and the stream forge; `**` is written `>` on the wire (modstreams.go +// WirePattern). The consumer's durable is _ on EVENTS (nats.go consumerDurable). The controller's own +// user keeps all rights, as it has. const busConf = ` listen: 127.0.0.1:-1 jetstream { store_dir: %q } @@ -31,7 +38,40 @@ accounts { jetstream: enabled users = [ { user: "ctl", password: "pw" } - { user: "n1.mesh-delivery-forge-bridge", password: "pw" } + { user: "n1.mesh-delivery-forge-bridge", password: "pw" + permissions { + publish { allow: [ + "$JS.ACK.EVENTS.n1_mesh-delivery-forge-bridge.>" + "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-forge-bridge_heads.>" + "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-forge-bridge_heads.>" + "$JS.API.CONSUMER.INFO.EVENTS.n1_mesh-delivery-forge-bridge" + "$JS.API.CONSUMER.MSG.NEXT.EVENTS.n1_mesh-delivery-forge-bridge" + "$JS.API.DIRECT.GET.KV_mesh-delivery-forge-bridge_heads.>" + "$JS.API.STREAM.INFO.KV_mesh-delivery-forge-bridge_heads" + "$JS.FC.KV_mesh-delivery-forge-bridge_heads.>" + "$KV.mesh-delivery-forge-bridge_heads.>" + "mesh.mod.gitea.call.gitea_commit_trees.*.n1~mesh-delivery-forge-bridge" + "mesh.mod.gitea.call.gitea_commit_trees.n1~mesh-delivery-forge-bridge" + "mesh.mod.gitea.call.gitea_open_pulls.*.n1~mesh-delivery-forge-bridge" + "mesh.mod.gitea.call.gitea_open_pulls.n1~mesh-delivery-forge-bridge" + "mesh.mod.gitea.tool.gitea_commit_trees" + "mesh.mod.gitea.tool.gitea_commit_trees.*" + "mesh.mod.gitea.tool.gitea_open_pulls" + "mesh.mod.gitea.tool.gitea_open_pulls.*" + "mesh.stream.mesh-delivery-forge-bridge.event.forge.>" + ] } + subscribe { allow: [ + "_INBOX.n1.mesh-delivery-forge-bridge.>" + "mesh.assignment.n1.mesh-delivery-forge-bridge" + "mesh.mod.gitea.event.pull.closed" + "mesh.mod.gitea.event.pull.merged" + "mesh.mod.gitea.event.pull.updated" + "mesh.mod.mesh-delivery-forge-bridge.call.>" + "mesh.mod.mesh-delivery-forge-bridge.tool.>" + ] } + allow_responses: true + } + } ] } } @@ -105,6 +145,10 @@ func startBed(t *testing.T, openReply string, prepare func(ctx context.Context, if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "FORGE", Subjects: []string{"mesh.stream.mesh-delivery-forge-bridge.event.forge.>"}}); err != nil { t.Fatal(err) } + // The bridge's state bucket, made by the controller from the manifest's state entry before the process starts. + if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: headsBucket, History: 1}); err != nil { + t.Fatal(err) + } if prepare != nil { prepare(ctx, js) } @@ -163,12 +207,13 @@ func TestBridgeWritesPullMessages(t *testing.T) { // 1235/AC2 } var info *jetstream.StreamInfo for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { - if info, err = forge.Info(ctx); err == nil && info.State.Msgs >= 1 { + if info, err = forge.Info(ctx); err == nil && info.State.Msgs >= 2 { break } } - if info.State.Msgs != 1 { - t.Fatalf("FORGE holds %d messages within 5 s, want 1", info.State.Msgs) + // The open message the bridge writes at its start (ADR 0443 §5), and the event's message. + if info.State.Msgs != 2 { + t.Fatalf("FORGE holds %d messages within 5 s, want 2", info.State.Msgs) } raw, err := forge.GetLastMsgForSubject(ctx, "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-delivery.pull") if err != nil { @@ -234,12 +279,13 @@ func TestBridgeWritesMergedMessage(t *testing.T) { // 1236/AC4 } var info *jetstream.StreamInfo for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { - if info, err = forge.Info(ctx); err == nil && info.State.Msgs >= 1 { + if info, err = forge.Info(ctx); err == nil && info.State.Msgs >= 2 { break } } - if info.State.Msgs != 1 { - t.Fatalf("FORGE holds %d messages within 5 s, want 1", info.State.Msgs) + // The open message the bridge writes at its start (ADR 0443 §5), and the event's message. + if info.State.Msgs != 2 { + t.Fatalf("FORGE holds %d messages within 5 s, want 2", info.State.Msgs) } raw, err := forge.GetLastMsgForSubject(ctx, "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-delivery.merged") if err != nil { @@ -324,8 +370,8 @@ func TestBridgeExitsWhenTheOpenPullsCannotBeRead(t *testing.T) { // 1237/AC3 } // TestBridgeReadsTheForgeAtStartAndAfterAReconnectAndHearsNoOlderEvent starts the bridge whole against a fake forge with two -// open pull requests, one of them with its head already on FORGE, and a bus whose EVENTS holds older pull events from -// before its consumer was made. At start it writes the one pull message that differs and the open message naming both; +// open pull requests, one of them with its head already kept in the bucket heads, and a bus whose EVENTS holds older pull events from +// before its consumer was made, under the permissions the controller composes for the bridge. At start it writes the one pull message that differs and the open message naming both; // the bus connection closed, it reads the forge again and writes the open message alone, since every head is held now; // for the older events it writes nothing (ADR 0443 §1, §5). func TestBridgeReadsTheForgeAtStartAndAfterAReconnectAndHearsNoOlderEvent(t *testing.T) { // 1279/AC8 @@ -348,9 +394,12 @@ func TestBridgeReadsTheForgeAtStartAndAfterAReconnectAndHearsNoOlderEvent(t *tes t.Fatal(err) } } - // What FORGE already holds: the head of mesh-delivery#12. - raw, _ := delivery.EncodeForge(delivery.Forge{Kind: delivery.ForgePull, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: head, Action: delivery.PullUpdated}) - if _, err := js.Publish(ctx, pullSubject("mesh-delivery"), raw); err != nil { + // What the bridge kept of an earlier write: the head of mesh-delivery#12. + kv, err := js.KeyValue(ctx, headsBucket) + if err != nil { + t.Fatal(err) + } + if _, err := kv.PutString(ctx, "novox.mesh-delivery.12", head); err != nil { t.Fatal(err) } }) @@ -388,12 +437,12 @@ func TestBridgeReadsTheForgeAtStartAndAfterAReconnectAndHearsNoOlderEvent(t *tes return len(got) == 2 && got[0] == bothOpen[0] && got[1] == bothOpen[1] } - // At start: the held head (message 1), the one that differs (2), the open message naming both (3). - waitMsgs(3) - if raw, err := forge.GetMsg(b.ctx, 2); err != nil || raw.Subject != pullSubject("mesh-catalog") { - t.Fatalf("message 2 = %v %v, want mesh-catalog's pull message", raw, err) + // At start: the one pull message that differs (1), the open message naming both (2). + waitMsgs(2) + if raw, err := forge.GetMsg(b.ctx, 1); err != nil || raw.Subject != pullSubject("mesh-catalog") { + t.Fatalf("message 1 = %v %v, want mesh-catalog's pull message", raw, err) } - if got := named(3); !same(got) { + if got := named(2); !same(got) { t.Fatalf("the open message names %v, want %v", got, bothOpen) } @@ -414,8 +463,8 @@ func TestBridgeReadsTheForgeAtStartAndAfterAReconnectAndHearsNoOlderEvent(t *tes if closed != 1 { t.Fatalf("%d connections of the bridge found", closed) } - waitMsgs(4) - if got := named(4); !same(got) { + waitMsgs(3) + if got := named(3); !same(got) { t.Fatalf("after the reconnect the open message names %v, want %v", got, bothOpen) } 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 e76b126..c067c25 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 @@ -188,8 +188,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro call := callerOn(conn, cred.User) // No forge event is heard while the open pull requests are read and said (ADR 0443 §5): events and syncs take turns. var turn sync.Mutex - held := func(ctx context.Context) (bridge.Held, error) { return heldOnForge(ctx, js) } - said := func(ctx context.Context) error { return syncOpen(ctx, call, js, held, &turn, logger) } + kv, err := js.KeyValue(ctx, headsBucket) + if err != nil { + return fmt.Errorf("the bucket %s: %w", headsBucket, err) + } + heads := kvHeads{kv} + // Everything written on FORGE, an event's and a sync's, keeps the head in the bucket. + rec := &recorder{pub: js, heads: heads, logger: logger} + said := func(ctx context.Context) error { return syncOpen(ctx, call, rec, heads, &turn, logger) } if err := said(ctx); err != nil { return err } @@ -198,7 +204,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return fmt.Errorf("the consumer %s_%s on %s: %w", cred.Node, cred.Module, eventsStream, err) } - cc, err := cons.Consume(func(m jetstream.Msg) { hear(&turn, func() { handle(ctx, m, js, treesVia(ctx, call), logger) }) }, + cc, err := cons.Consume(func(m jetstream.Msg) { + hear(&turn, func() { + // 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), logger) + }) + }, jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) })) if err != nil { return err diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/open.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/open.go index 8a9c576..f09913f 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/open.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/open.go @@ -4,7 +4,13 @@ import ( "context" "fmt" "log" + "slices" "sync" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" "mesh-delivery-forge-bridge/internal/bridge" ) @@ -13,8 +19,50 @@ import ( // gitea_open_pulls: no arguments, answers {pulls: [...]}). const openPullsTool = "gitea.gitea_open_pulls" -// heldReader reads the head FORGE last holds for each pull request. -type heldReader func(ctx context.Context) (bridge.Held, error) +// headStore is the bucket heads: the head last written on FORGE for each pull request, by bridge.Track's key +// (novox/hq ADR 0443 §5). +type headStore interface { + // Get is the head kept under the key; found is false when there is none. + Get(ctx context.Context, key string) (head string, found bool, err error) + Put(ctx context.Context, key, head string) error + Delete(ctx context.Context, key string) error +} + +// recorder writes to FORGE and keeps in the bucket what it wrote: a pull message keeps its head, a closed or merged one +// takes its pull request's key away. Every write of the program goes through it, an event's and a sync's alike. +type recorder struct { + pub publisher + heads headStore + logger *log.Logger +} + +func (r *recorder) Publish(ctx context.Context, subject string, payload []byte, opts ...jetstream.PublishOpt) (*jetstream.PubAck, error) { + ack, err := r.pub.Publish(ctx, subject, payload, opts...) + if err != nil { + return ack, err + } + f, derr := delivery.DecodeForge(payload) + if derr != nil { + return ack, nil + } + key, head, ok := bridge.Track(f) + if !ok { + return ack, nil + } + if head == "" { + err = r.heads.Delete(ctx, key) + } else { + err = r.heads.Put(ctx, key, head) + } + if err != nil { + return ack, fmt.Errorf("%s was written on FORGE but its head could not be kept in the bucket: %w", subject, err) + } + return ack, nil +} + +// syncDeadline bounds one sync: the call to the forge and every read and write of the bucket and of FORGE, because +// the sync holds the turn that stops the bridge hearing any event. +const syncDeadline = 30 * time.Second // hear runs one reaction to a forge event while no sync of the open pull requests is between reading the forge and // writing the open message (novox/hq ADR 0443 §5): an event heard meanwhile waits, and is written after it, so a pull @@ -25,15 +73,17 @@ func hear(mu *sync.Mutex, react func()) { react() } -// syncOpen reads the open pull requests from the forge and writes a pull message for each whose head FORGE does not -// hold, then one open message naming every one it read, those whose head did not differ among them: what makes the -// indexer's candidates match the forge after a restart or a gap in the forge's events (ADR 0429 §13, ADR 0443 §5). It -// holds mu from before it reads the forge until the open message is written, so no event is heard between. What FORGE -// holds that cannot be read is logged, and every open pull request is written. An error names the call and the forge's -// words. -func syncOpen(ctx context.Context, call caller, pub publisher, held heldReader, mu *sync.Mutex, logger *log.Logger) error { +// syncOpen reads the open pull requests from the forge and writes a pull message for each whose head differs from the +// one kept in the bucket, then one open message naming every one it read, those whose head did not differ among them: +// what makes the indexer's candidates match the forge after a restart or a gap in the forge's events (ADR 0429 §13, +// ADR 0443 §5). pub keeps the bucket as it writes (recorder). It holds mu from before it reads the forge until the open +// message is written, so no event is heard between, and never past syncDeadline. A bucket that cannot be read is logged +// and every open pull request is written. An error names the call and the forge's words. +func syncOpen(ctx context.Context, call caller, pub publisher, heads headStore, mu *sync.Mutex, logger *log.Logger) error { mu.Lock() defer mu.Unlock() + ctx, cancel := context.WithTimeout(ctx, syncDeadline) + defer cancel() raw, err := call(ctx, openPullsTool, struct{}{}) if err != nil { return fmt.Errorf("%s: %w", openPullsTool, err) @@ -49,17 +99,30 @@ func syncOpen(ctx context.Context, call caller, pub publisher, held heldReader, for _, s := range skipped { logger.Printf("an open pull request is left out: %s", s) } - if h, err := held(ctx); err != nil { - logger.Printf("what FORGE holds could not be read, so every open pull request is written: %v", err) - } else { - msgs = h.Differing(msgs) + var differing []bridge.Message + for _, m := range msgs { + f, err := delivery.DecodeForge(m.Data) + if err != nil { + differing = append(differing, m) + continue + } + key, head, _ := bridge.Track(f) + kept, found, err := heads.Get(ctx, key) + if err != nil { + logger.Printf("the heads kept could not be read, so every open pull request is written: %v", err) + differing = slices.Clone(msgs) + break + } + if !found || kept != head { + differing = append(differing, m) + } } - for _, m := range append(msgs, open) { + for _, m := range append(differing, open) { if _, err := pub.Publish(ctx, forgeSubject(m.Event), m.Data); err != nil { return fmt.Errorf("the open pull request message %s could not be written on FORGE: %w", m.Event, err) } } - logger.Printf("%d pull messages and the open message written on FORGE", len(msgs)) + logger.Printf("%d pull messages and the open message written on FORGE", len(differing)) return nil } diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/open_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/open_test.go index 62d5b70..c1009b3 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/open_test.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/open_test.go @@ -14,8 +14,6 @@ import ( "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/delivery" - - "mesh-delivery-forge-bridge/internal/bridge" ) // openAnswer is gitea_open_pulls's answer as mesh-catalog modules/gitea/client.ts:369-381 builds its entries inside {pulls}. @@ -34,7 +32,7 @@ func TestSyncOpenWritesOneUpdatedMessagePerOpenPull(t *testing.T) { // 1237/AC1, keys = append(keys, key) return openAnswer(), nil } - if err := syncOpen(context.Background(), call, pub, nothingHeld, &sync.Mutex{}, log.New(&logs, "", 0)); err != nil { + if err := runSync(call, pub, newHeads(), &sync.Mutex{}, log.New(&logs, "", 0)); err != nil { t.Fatal(err) } if len(keys) != 1 || keys[0] != "gitea.gitea_open_pulls" { @@ -52,22 +50,59 @@ func TestSyncOpenWritesOneUpdatedMessagePerOpenPull(t *testing.T) { // 1237/AC1, } } -func nothingHeld(context.Context) (bridge.Held, error) { return bridge.Held{}, nil } +// memHeads is the bucket heads in memory. +type memHeads struct { + mu sync.Mutex + vals map[string]string + getErr error +} + +func newHeads(kv ...string) *memHeads { + m := &memHeads{vals: map[string]string{}} + for i := 0; i+1 < len(kv); i += 2 { + m.vals[kv[i]] = kv[i+1] + } + return m +} + +func (m *memHeads) Get(_ context.Context, key string) (string, bool, error) { + m.mu.Lock() + defer m.mu.Unlock() + v, ok := m.vals[key] + return v, ok, m.getErr +} +func (m *memHeads) Put(_ context.Context, key, head string) error { + m.mu.Lock() + defer m.mu.Unlock() + m.vals[key] = head + return nil +} +func (m *memHeads) Delete(_ context.Context, key string) error { + m.mu.Lock() + defer m.mu.Unlock() + delete(m.vals, key) + return nil +} +func (m *memHeads) get(key string) string { m.mu.Lock(); defer m.mu.Unlock(); return m.vals[key] } + +// runSync runs syncOpen with the recorder the program writes through. +func runSync(call caller, pub publisher, heads headStore, mu *sync.Mutex, logger *log.Logger) error { + return syncOpen(context.Background(), call, &recorder{pub: pub, heads: heads, logger: logger}, heads, mu, logger) +} // 1279/AC6: the sync writes a pull message only for a head that differs from what FORGE holds, then one open message // naming every open pull request into trunk it read, those whose head did not differ among them. func TestSyncOpenWritesPullsForNewHeadsThenOneOpenMessageNamingAll(t *testing.T) { // 1279/AC6 pub := &fakePub{} call := func(context.Context, string, any) (json.RawMessage, error) { return openAnswer(), nil } - held := func(context.Context) (bridge.Held, error) { - h := bridge.Held{} - h.Hear(delivery.Forge{Kind: delivery.ForgePull, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: head, Action: delivery.PullUpdated}) - return h, nil - } + heads := newHeads("novox.mesh-delivery.12", head) var logs bytes.Buffer - if err := syncOpen(context.Background(), call, pub, held, &sync.Mutex{}, log.New(&logs, "", 0)); err != nil { + if err := runSync(call, pub, heads, &sync.Mutex{}, log.New(&logs, "", 0)); err != nil { t.Fatal(err) } + if heads.get("novox.mesh-catalog.300") != strings.Repeat("a", 40) || heads.get("novox.mesh-delivery.12") != head { + t.Errorf("the bucket after the sync: %v, want the head written for mesh-catalog#300 kept beside the held one", heads.vals) + } if len(pub.subjects) != 2 { t.Fatalf("%d messages written (%v), want the one pull message that differs and the open message", len(pub.subjects), pub.subjects) } @@ -84,21 +119,59 @@ func TestSyncOpenWritesPullsForNewHeadsThenOneOpenMessageNamingAll(t *testing.T) } } -// What FORGE holds cannot be read: every open pull request is written, as before, and the open message follows; the +// The bucket cannot be read: every open pull request is written, as before, and the open message follows; the // indexer's writes are idempotent, so the fault costs messages and no state. -func TestSyncOpenWritesAllWhenFORGECannotBeRead(t *testing.T) { // 1279/AC6 +func TestSyncOpenWritesAllWhenTheHeadsCannotBeRead(t *testing.T) { // 1279/AC6 pub := &fakePub{} call := func(context.Context, string, any) (json.RawMessage, error) { return openAnswer(), nil } - held := func(context.Context) (bridge.Held, error) { return nil, errors.New("no consumer on FORGE") } + heads := newHeads("novox.mesh-delivery.12", head) + heads.getErr = errors.New("the bucket is gone") var logs bytes.Buffer - if err := syncOpen(context.Background(), call, pub, held, &sync.Mutex{}, log.New(&logs, "", 0)); err != nil { + if err := runSync(call, pub, heads, &sync.Mutex{}, log.New(&logs, "", 0)); err != nil { t.Fatal(err) } - if len(pub.subjects) != 3 || !strings.Contains(logs.String(), "no consumer on FORGE") { + if len(pub.subjects) != 3 || !strings.Contains(logs.String(), "the bucket is gone") { t.Errorf("messages %v, log %q", pub.subjects, logs.String()) } } +// What the bridge writes on FORGE it keeps in the bucket: a head by its pull message, nothing by a closed or merged one. +func TestWhatIsWrittenOnFORGEIsKeptInTheBucket(t *testing.T) { // 1279/AC6 + pub := &fakePub{} + heads := newHeads() + r := &recorder{pub: pub, heads: heads, logger: log.New(&bytes.Buffer{}, "", 0)} + write := func(f delivery.Forge) { + t.Helper() + ev, _ := delivery.ForgeEvent(f.Repository, f.Kind) + raw, err := delivery.EncodeForge(f) + if err != nil { + t.Fatal(err) + } + if _, err := r.Publish(context.Background(), forgeSubject(ev), raw); err != nil { + t.Fatal(err) + } + } + p := delivery.Forge{Kind: delivery.ForgePull, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: head, Action: delivery.PullUpdated} + write(p) + if heads.get("novox.mesh-delivery.12") != head { + t.Fatalf("bucket %v", heads.vals) + } + p.Action = delivery.PullClosed + write(p) + if _, kept := heads.vals["novox.mesh-delivery.12"]; kept { + t.Errorf("a closed pull request is still kept: %v", heads.vals) + } + p.Action = delivery.PullUpdated + write(p) + write(delivery.Forge{Kind: delivery.ForgeMerged, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: head, Merge: "m", Root: "r", Trees: map[string]string{"": "t"}}) + if _, kept := heads.vals["novox.mesh-delivery.12"]; kept { + t.Errorf("a merged pull request is still kept: %v", heads.vals) + } + if len(pub.subjects) != 4 { + t.Errorf("%d messages written", len(pub.subjects)) + } +} + // 1279/AC6: no forge event is heard between reading the forge and writing the open message. func TestNoForgeEventIsHeardWhileTheOpenPullsAreSaid(t *testing.T) { // 1279/AC6 pub := &lockedPub{} @@ -112,7 +185,7 @@ func TestNoForgeEventIsHeardWhileTheOpenPullsAreSaid(t *testing.T) { // 1279/AC6 var mu sync.Mutex synced := make(chan error, 1) go func() { - synced <- syncOpen(context.Background(), call, pub, nothingHeld, &mu, log.New(&bytes.Buffer{}, "", 0)) + synced <- runSync(call, pub, newHeads(), &mu, log.New(&bytes.Buffer{}, "", 0)) }() <-reading // An event arrives while the forge is being read. @@ -158,7 +231,7 @@ func TestSyncOpenFailureNamesTheCallAndTheForgesWords(t *testing.T) { // 1237/AC call := func(context.Context, string, any) (json.RawMessage, error) { return nil, errors.New("the forge said 502") } - err := syncOpen(context.Background(), call, pub, nothingHeld, &sync.Mutex{}, log.New(&bytes.Buffer{}, "", 0)) + err := runSync(call, pub, newHeads(), &sync.Mutex{}, log.New(&bytes.Buffer{}, "", 0)) if err == nil || !strings.Contains(err.Error(), "gitea.gitea_open_pulls") || !strings.Contains(err.Error(), "the forge said 502") { t.Errorf("err = %v", err) } diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go index 7a3ebed..e845d91 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go @@ -7,6 +7,7 @@ import ( "fmt" "regexp" "slices" + "strconv" "strings" "git.novox.be/novox/mesh-sdk/go/delivery" @@ -189,34 +190,18 @@ func OpenPulls(answer []byte) (msgs []Message, skipped []string, err error) { return msgs, skipped, nil } -// Held is the head FORGE last holds for each pull request, by `/#`: what the bridge compares an -// open pull request's head with, so it writes a pull message only for a head that differs (novox/hq ADR 0443 §5). -type Held map[string]string - -func heldKey(repository string, number int) string { return fmt.Sprintf("%s#%d", repository, number) } - -// Hear takes one message of FORGE into what is held: a pull message holds its head, a closed one or a merged message -// holds nothing for its pull request. Other kinds change nothing. -func (h Held) Hear(f delivery.Forge) { - key := heldKey(f.Repository, f.Number) +// Track says what a message the bridge wrote changes in the head it keeps for each pull request (the bucket heads, +// novox/hq ADR 0443 §5): the key `..`, and the head to keep under it, or "" to take the key away +// (a closed pull request, a merged one). ok is false for a message that keeps no head. +func Track(f delivery.Forge) (key, head string, ok bool) { + key = strings.ReplaceAll(f.Repository, "/", ".") + "." + strconv.Itoa(f.Number) switch { case f.Kind == delivery.ForgePull && f.Action != delivery.PullClosed: - h[key] = f.Head + return key, f.Head, true case f.Kind == delivery.ForgePull || f.Kind == delivery.ForgeMerged: - delete(h, key) + return key, "", true } -} - -// Differing keeps the pull messages whose head FORGE does not hold, in their order. -func (h Held) Differing(msgs []Message) []Message { - var out []Message - for _, m := range msgs { - if f, err := delivery.DecodeForge(m.Data); err == nil && h[heldKey(f.Repository, f.Number)] == f.Head { - continue - } - out = append(out, m) - } - return out + return "", "", false } // OpenMessage turns the answer of gitea_open_pulls into the open message: every open pull request into the trunk, by diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge_test.go b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge_test.go index 507ad48..60a7ca5 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge_test.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge_test.go @@ -350,31 +350,25 @@ func TestOpenMessageNamesEveryOpenPullIntoTheTrunk(t *testing.T) { // 1279/AC6 } } -func TestOnlyAHeadFORGEDoesNotHoldIsWritten(t *testing.T) { // 1279/AC6 - h2 := strings.Repeat("a", 40) - msgs, _, err := OpenPulls(openAnswer(openEntry("mesh-delivery", 12, "main", "feat/x", head), openEntry("mesh-catalog", 300, "main", "feat/x", h2))) - if err != nil { - t.Fatal(err) +func TestTrackSaysWhichHeadOfWhichPullRequestWasWritten(t *testing.T) { // 1279/AC6 + pullMsg := func(action string) delivery.Forge { + return delivery.Forge{Kind: delivery.ForgePull, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: head, Action: action} } - held := Held{} - held.Hear(delivery.Forge{Kind: delivery.ForgePull, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: head, Action: delivery.PullUpdated}) - got := held.Differing(msgs) - if len(got) != 1 || got[0].Event != "forge.novox.mesh-catalog.pull" { - t.Fatalf("written: %+v, want only mesh-catalog#300", got) + for _, a := range []string{delivery.PullOpened, delivery.PullUpdated} { + if key, h, ok := Track(pullMsg(a)); key != "novox.mesh-delivery.12" || h != head || !ok { + t.Errorf("%s: Track = %q %q %v, want the head kept under a bucket key", a, key, h, ok) + } } - // Another head for the same pull request differs; a closed or merged one is no longer held. - held.Hear(delivery.Forge{Kind: delivery.ForgePull, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: h2, Action: delivery.PullUpdated}) - if got := held.Differing(msgs[:1]); len(got) != 1 { - t.Errorf("a pull request held at another head was left out") + // A closed pull request and a merged one are no longer kept: their key goes. + if key, h, ok := Track(pullMsg(delivery.PullClosed)); key != "novox.mesh-delivery.12" || h != "" || !ok { + t.Errorf("closed: Track = %q %q %v", key, h, ok) } - held.Hear(delivery.Forge{Kind: delivery.ForgePull, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: head, Action: delivery.PullUpdated}) - held.Hear(delivery.Forge{Kind: delivery.ForgePull, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: head, Action: delivery.PullClosed}) - if got := held.Differing(msgs[:1]); len(got) != 1 { - t.Errorf("a closed pull request is still held") + merged := delivery.Forge{Kind: delivery.ForgeMerged, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: head, Merge: "m", Root: "r", Trees: map[string]string{"": "t"}} + if key, h, ok := Track(merged); key != "novox.mesh-delivery.12" || h != "" || !ok { + t.Errorf("merged: Track = %q %q %v", key, h, ok) } - held.Hear(delivery.Forge{Kind: delivery.ForgePull, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: head, Action: delivery.PullOpened}) - held.Hear(delivery.Forge{Kind: delivery.ForgeMerged, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: head, Merge: "m", Root: "r", Trees: map[string]string{"": "t"}}) - if got := held.Differing(msgs[:1]); len(got) != 1 { - t.Errorf("a merged pull request is still held") + // The open message and the rest hold no head. + if _, _, ok := Track(delivery.Forge{Kind: delivery.ForgeOpen}); ok { + t.Error("an open message was tracked") } } diff --git a/modules/mesh-delivery-forge-bridge/manifest_test.go b/modules/mesh-delivery-forge-bridge/manifest_test.go index 6cac408..04d3557 100644 --- a/modules/mesh-delivery-forge-bridge/manifest_test.go +++ b/modules/mesh-delivery-forge-bridge/manifest_test.go @@ -22,6 +22,11 @@ type stream struct { Events []string `json:"events"` } +type state struct { + Name string `json:"name"` + History int `json:"history"` +} + type artifact struct { Name string `json:"name"` Kind string `json:"kind"` @@ -37,6 +42,7 @@ type manifest struct { Slug string `json:"slug"` Consumes []string `json:"consumes"` Streams []stream `json:"streams"` + State []state `json:"state"` Reads []string `json:"reads"` Invokes []string `json:"invokes"` OwnSecrets map[string]string `json:"own-secrets"` @@ -96,8 +102,12 @@ func TestManifestDeclarations(t *testing.T) { // 1235/AC1 if m.SecretsOwner != "mesh-delivery-forge-bridge" { t.Errorf("secrets-owner = %q", m.SecretsOwner) } - // No state, no bus, no seat, no tool (ADR 0437 §4). - for _, k := range []string{"state", "logs", "emits", "claims", "provides", "requires", "uses", "defines-seats", "tools", "seat", "seats"} { + // One bucket: the head last written on FORGE for each pull request, so a sync writes only what differs (ADR 0443 §5). + if !reflect.DeepEqual(m.State, []state{{Name: "heads", History: 1}}) { // 1279/AC6 + t.Errorf("state = %+v, want the one bucket heads", m.State) + } + // No other state, no bus, no seat, no tool (ADR 0437 §4). + for _, k := range []string{"logs", "emits", "claims", "provides", "requires", "uses", "defines-seats", "tools", "seat", "seats"} { if _, ok := top[k]; ok { t.Errorf("the manifest declares %q: the bridge has none", k) } diff --git a/modules/mesh-delivery-forge-bridge/module.json b/modules/mesh-delivery-forge-bridge/module.json index ee6e568..ec78755 100644 --- a/modules/mesh-delivery-forge-bridge/module.json +++ b/modules/mesh-delivery-forge-bridge/module.json @@ -10,6 +10,9 @@ "streams": [ { "name": "forge", "events": ["forge.**"] } ], + "state": [ + { "name": "heads", "history": 1 } + ], "reads": [ "mesh-delivery-gate.merges", "mesh-delivery-gate.owed"