From dce4d01a584687181851608355b26fa2a2d3fb8a Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 09:15:04 +0100 Subject: [PATCH] FB1, FB9, FB10: the forge bridge writes one message per move of every watched branch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The bridge watches the registrations' branches by origin (a first view of registrations, then registration-changed), before the first registration the gate's source's (source-moved); no branch is named in its code, Trunk goes. Every move is heard on the seat git's branch-moved through the bridge's durable on SEAT_git_branch-moved, its head read by the seat's verb branch, never the event's: merged with its pull request and `base` (the watched branch) when the seat's pull names the head a pull request's merge, else `pushed` with branch, head and `before`, both with the take-in inputs at the head. The last head written per watched branch is kept in the state heads under branch..., so a move heard twice writes once, and each watched branch's head is read at the start and when a branch is registered. A move of an unprotected branch writes nothing. pull.merged into a watched branch writes nothing (its message is the move's); the forge-made merge's trunk filter goes. Builds: design 47 G §4, §7, §11, §20, §41, §507, §509, §532, §714, A-227, A-236 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../mesh-delivery-forge-bridge-tools/main.go | 2 +- .../main_test.go | 10 +- .../integration_test.go | 66 +++++--- .../cmd/mesh-delivery-forge-bridge/main.go | 139 ++++++++++++----- .../mesh-delivery-forge-bridge/main_test.go | 100 ++++++++---- .../cmd/mesh-delivery-forge-bridge/moves.go | 143 ++++++++++++++++++ .../cmd/mesh-delivery-forge-bridge/open.go | 6 +- .../mesh-delivery-forge-bridge/open_test.go | 2 +- .../forgemsg/forge.go | 65 +++++--- .../forgemsg/forge_test.go | 9 +- modules/mesh-delivery-forge-bridge/go.mod | 2 + .../internal/bridge/bridge.go | 102 ++++++------- .../internal/bridge/bridge_test.go | 87 +++++++---- .../internal/bridge/call.go | 19 ++- .../internal/bridge/files_test.go | 36 ++--- .../internal/bridge/forgemade.go | 13 +- .../internal/bridge/moved.go | 95 ++++++++++++ .../internal/bridge/moved_test.go | 77 ++++++++++ .../internal/bridge/paths_test.go | 2 +- .../internal/bridge/watched.go | 71 +++++++++ .../manifest_test.go | 8 +- .../mesh-delivery-forge-bridge/module.json | 6 + 22 files changed, 819 insertions(+), 241 deletions(-) create mode 100644 modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/moves.go create mode 100644 modules/mesh-delivery-forge-bridge/internal/bridge/moved.go create mode 100644 modules/mesh-delivery-forge-bridge/internal/bridge/moved_test.go create mode 100644 modules/mesh-delivery-forge-bridge/internal/bridge/watched.go diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main.go index 6a381e25..31ab8967 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main.go @@ -131,7 +131,7 @@ func MergedOf(of []seat.Group, repository, merge string, number int, trees bridg if err != nil { return nil, fmt.Errorf("the forge gives no merged pull request of %s with the commit %s: %w", repository, merge, err) } - msg, err := bridge.MergedMessage(owner, repo, fp.Number, fp.Branch, fp.Head, merge, trees, files) + msg, err := bridge.MergedMessage(owner, repo, fp.Number, fp.Branch, fp.Head, merge, fp.Base, trees, files) if err != nil { return nil, fmt.Errorf("the merged message of %s at %s cannot be built: %w", repository, merge, err) } diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main_test.go index c39fecb5..83d8a758 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main_test.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main_test.go @@ -135,8 +135,7 @@ func TestTheBridgeToolsReadTheGroupsAndAskTheEndAgain(t *testing.T) { return nil, bridge.ErrNotFound } published := func(repo string, number int, head, branch, merge string) forgemsg.Forge { - body, _ := json.Marshal(git.PullMergedEvent{Repository: "novox/" + repo, Number: int64(number), Branch: branch, Head: head, Base: "main", MergeCommit: merge}) - msg, err := bridge.Convert(bridge.MergedSubject, body, trees, files) + msg, err := bridge.MergedMessage("novox", repo, number, branch, head, merge, "main", trees, files) if err != nil { t.Fatal(err) } @@ -172,8 +171,13 @@ func TestTheBridgeToolsReadTheGroupsAndAskTheEndAgain(t *testing.T) { api: forgeAnswers(t, 9, "", "merged", "fix/x", h2, c1, "main"), fail: "not at " + c2}, {name: "a forge pull request not merged", repository: "novox/a", merge: c2, number: 9, api: forgeAnswers(t, 9, "", "open", "fix/x", h2, "", "main"), fail: "not merged"}, + // design 47 G §714: no filter to the trunk alone; the merged message names the branch it was merged into. {name: "a forge pull request merged into another branch", repository: "novox/a", merge: c2, number: 9, - api: forgeAnswers(t, 9, "", "merged", "fix/x", h2, c2, "dev"), fail: "not the trunk"}, + api: forgeAnswers(t, 9, "", "merged", "fix/x", h2, c2, "dev"), want: func() forgemsg.Forge { + f := published("a", 9, h2, "fix/x", c2) + f.Base = "dev" + return f + }()}, {name: "a forge pull request with no head branch", repository: "novox/a", merge: c2, number: 9, api: forgeAnswers(t, 9, "", "merged", "", h2, c2, "main"), fail: "no head branch"}, {name: "a seat answer that says an error", repository: "novox/a", merge: c2, number: 9, 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 34fa9523..6f658a15 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 @@ -97,6 +97,16 @@ accounts { "$JS.API.CONSUMER.MSG.NEXT.SEAT_mesh-delivery-checker_owed.n1_mesh-delivery-forge-bridge" "mesh.seat.mesh-broker.call.show.n1~mesh-delivery-forge-bridge" "mesh.seat.mesh-broker.tool.show" + "mesh.seat.mesh-catalog.call.registrations.n1~mesh-delivery-forge-bridge" + "mesh.seat.mesh-catalog.tool.registrations" + "mesh.seat.mesh-delivery.call.source.n1~mesh-delivery-forge-bridge" + "mesh.seat.mesh-delivery.tool.source" + "mesh.seat.git.call.branch.n1~mesh-delivery-forge-bridge" + "mesh.seat.git.tool.branch" + "$JS.API.STREAM.INFO.SEAT_git_branch-moved" + "$JS.ACK.SEAT_git_branch-moved.n1_mesh-delivery-forge-bridge.>" + "$JS.API.CONSUMER.INFO.SEAT_git_branch-moved.n1_mesh-delivery-forge-bridge" + "$JS.API.CONSUMER.MSG.NEXT.SEAT_git_branch-moved.n1_mesh-delivery-forge-bridge" "mesh.seat.git.call.status.n1~mesh-delivery-forge-bridge" "mesh.seat.git.tool.status" "mesh.seat.git.call.comment.n1~mesh-delivery-forge-bridge" @@ -232,7 +242,7 @@ func startBed(t *testing.T, openReply string, prepare func(ctx context.Context, } // Each seat's own stream owed, as its contract keeps it (Stream, SEAT__owed), as the grants part makes it: no // subjects of its own, sourced from EVENTS filtered by the event's subject. - for name, subjects := range map[string]string{seat.StreamOwed: seat.SubjectOwedAll, checker.StreamOwed: checker.SubjectOwedAll} { + for name, subjects := range map[string]string{seat.StreamOwed: seat.SubjectOwedAll, checker.StreamOwed: checker.SubjectOwedAll, git.StreamBranchMoved: git.SubjectBranchMoved} { if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: name, MaxAge: 7 * 24 * time.Hour, Sources: []*jetstream.StreamSource{{Name: "EVENTS", FilterSubject: subjects}}}); err != nil { t.Fatal(err) @@ -251,11 +261,12 @@ func startBed(t *testing.T, openReply string, prepare func(ctx context.Context, acct := process.Account{Node: "n1", Module: "mesh-delivery-forge-bridge"} if _, err := js.CreateOrUpdateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{ Durable: acct.EventConsumerName(), AckPolicy: jetstream.AckExplicitPolicy, DeliverPolicy: jetstream.DeliverNewPolicy, - FilterSubjects: []string{git.SubjectPullUpdated, git.SubjectPullMerged, git.SubjectPullClosed, git.SubjectForgeDelete}, + FilterSubjects: []string{git.SubjectPullUpdated, git.SubjectPullMerged, git.SubjectPullClosed, git.SubjectForgeDelete, + "mesh.seat.mesh-catalog.event.registration-changed", "mesh.seat.mesh-delivery.event.source-moved"}, }); err != nil { t.Fatal(err) } - for name, subjects := range map[string]string{seat.StreamOwed: seat.SubjectOwedAll, checker.StreamOwed: checker.SubjectOwedAll} { + for name, subjects := range map[string]string{seat.StreamOwed: seat.SubjectOwedAll, checker.StreamOwed: checker.SubjectOwedAll, git.StreamBranchMoved: git.SubjectBranchMoved} { if _, err := js.CreateOrUpdateConsumer(ctx, name, acct.SeatEventConsumerConfig([]string{subjects}, false)); err != nil { t.Fatal(err) } @@ -266,6 +277,9 @@ func startBed(t *testing.T, openReply string, prepare func(ctx context.Context, answers := map[string]string{ "git.open-pulls": openReply, "git.trees": `{"root":"r0","trees":{"":"t0","modules/a":"t1"}}`, "mesh-delivery.groups": `{"groups":[]}`, "mesh-hand-acts.hand-act": `{"recorded":true}`, + // No registration yet: the watched branches are the gate's source's, main (design 47 G §507). + "mesh-catalog.registrations": `{"registrations":[]}`, "mesh-delivery.source": `{"branches":["main"]}`, + "git.pull": `{"error":"no pull request merged at that commit","error-kind":"not-found"}`, "mesh-broker.show": `{"state":"created","binding":{"local":"performed","storage":"key-value","template":"state","bucket":"` + boundPerformed + `"}}`, } for _, verb := range []string{"status", "comment", "view", "note", "merge-pull"} { @@ -280,6 +294,10 @@ func startBed(t *testing.T, openReply string, prepare func(ctx context.Context, mu.Lock() called = append(called, verb+" "+string(m.Data)) answer, files := answers[verb], filesOf + var in struct{ Repository string } + if json.Unmarshal(m.Data, &in) == nil && answers[verb+" "+in.Repository] != "" { + answer = answers[verb+" "+in.Repository] // a move's branch and pull, set per repository (landedOn) + } mu.Unlock() if verb == "git.file" { var in git.FileInput @@ -326,6 +344,23 @@ func startBed(t *testing.T, openReply string, prepare func(ctx context.Context, }} } +// landedOn sets the stand-in seat git so main of repository is protected at at, which the pull request number on branch +// at prHead merged. +func landedOn(b *bed, repository string, number int, branch, prHead, at string) { + b.answer("git.branch "+repository, `{"repository":"`+repository+`","branch":"main","head":"`+at+`","protected":true}`) + b.answer("git.pull "+repository, fmt.Sprintf(`{"pull":{"repository":%q,"number":%d,"branch":%q,"head":%q,"base":"main","state":"merged","merge-commit":%q}}`, + repository, number, branch, prHead, at)) +} + +// moveMain says main of repository moved, as the seat git says it, naming a head the bridge never takes (§11). +func moveMain(t *testing.T, b *bed, repository string) { + t.Helper() + ev, _ := json.Marshal(git.BranchMovedEvent{Repository: repository, Branch: "main", Head: strings.Repeat("f", 40)}) + if _, err := b.js.Publish(b.ctx, git.SubjectBranchMoved, ev); err != nil { + t.Fatal(err) + } +} + // TestBridgeWritesPullMessages starts the program the way the node does (its entry, run, with MESH_BROKER_FILE). func TestBridgeWritesPullMessages(t *testing.T) { // 1235/AC2 ctl, js, ctx, _ := startBridge(t, `{"pulls":[]}`) @@ -452,12 +487,9 @@ func TestBridgeWritesMergedMessage(t *testing.T) { // 1236/AC4, 1448/AC5 } return "", false }) - // pull.merged as the seat git says it (its contract's PullMergedEvent). - ev, _ := json.Marshal(git.PullMergedEvent{Repository: "novox/mesh-delivery", Number: 12, Title: "t", Head: head, Branch: "feat/x", Base: "main", - MergeCommit: merge, Change: &git.Change{Paths: []string{"modules/a/x.go"}, ModuleDirs: []string{"modules/a"}}}) - if _, err := js.Publish(ctx, git.SubjectPullMerged, ev); err != nil { - t.Fatal(err) - } + // design 47 G §11, §509, §714: main moved, read by the seat's branch, and the forge names #12 merged at its head. + landedOn(bed, "novox/mesh-delivery", 12, "feat/x", head, merge) + moveMain(t, bed, "novox/mesh-delivery") forge, err := js.Stream(ctx, "FORGE") if err != nil { @@ -482,7 +514,7 @@ func TestBridgeWritesMergedMessage(t *testing.T) { // 1236/AC4, 1448/AC5 t.Fatal(err) } want := forgemsg.Forge{V: keys.Version, Kind: "merged", Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, Head: head, - Merge: merge, Root: "r0", Trees: map[string]string{"": "t0", "modules/a": "t1"}, + Merge: merge, Base: "main", Root: "r0", Trees: map[string]string{"": "t0", "modules/a": "t1"}, Manifests: map[string]string{"": string(manifest), "modules/a": string(manifest)}, // No mesh package: only the indexer computes what a fingerprint takes (ADR 0593 §2). Locks: map[string]map[string]string{"modules/a": {"go.sum": hex.EncodeToString(sha256Sum(sum))}}} @@ -499,7 +531,7 @@ func TestBridgeWritesMergedMessage(t *testing.T) { // 1236/AC4, 1448/AC5 if len(trees) != 1 || trees[0] != "git.trees "+string(in) { t.Errorf("the seat git was asked for trees %v, want once at the merge commit", trees) } - cons, err := js.Consumer(ctx, "EVENTS", "n1_mesh-delivery-forge-bridge") + cons, err := js.Consumer(ctx, git.StreamBranchMoved, "n1_mesh-delivery-forge-bridge") if err != nil { t.Fatal(err) } @@ -725,14 +757,10 @@ func TestBridgeSaysAGroupFromTheSeatsGroups(t *testing.T) { // 1585, ADR 0633 § {Repository: "novox/mesh-delivery", Number: 12, Head: head, State: "merged"}, {Repository: "novox/mesh-sdk", Number: 4, Head: headB, State: "merging"}}}}}) b.answer("mesh-delivery.groups", string(groups)) ch := heard(t, b) - for _, m := range []git.PullMergedEvent{ - {Repository: "novox/mesh-delivery", Number: 12, Head: head, Branch: "feat/x", Base: "main", MergeCommit: merge, Change: &git.Change{Paths: []string{"modules/a/x.go"}, ModuleDirs: []string{"modules/a"}}}, - {Repository: "novox/mesh-sdk", Number: 4, Head: headB, Branch: "feat/x", Base: "main", MergeCommit: mergeB, Change: &git.Change{Paths: []string{"go/x.go"}}}, - } { - ev, _ := json.Marshal(m) - if _, err := b.js.Publish(b.ctx, git.SubjectPullMerged, ev); err != nil { - t.Fatal(err) - } + landedOn(b, "novox/mesh-delivery", 12, "feat/x", head, merge) + landedOn(b, "novox/mesh-sdk", 4, "feat/x", headB, mergeB) + for _, m := range []git.PullMergedEvent{{Repository: "novox/mesh-delivery"}, {Repository: "novox/mesh-sdk"}} { + moveMain(t, b, m.Repository) repo := strings.TrimPrefix(m.Repository, "novox/") f, err := forgemsg.DecodeForge(next(t, ch, "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox."+repo+".merged")) if err != nil || f.Group != "feat/x" || len(f.Members) != 2 || f.Members[1].Number != 4 { 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 91ff4208..0adb3881 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 @@ -6,6 +6,7 @@ package main import ( "context" + "encoding/json" "errors" "fmt" "io" @@ -17,14 +18,17 @@ import ( "time" "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/views" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" git "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/git" + mcseat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery" checker "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdeliverychecker" - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/sourcebranches" "mesh-delivery-forge-bridge/internal/bridge" groupkeys "mesh-delivery-forge-bridge/internal/groups" @@ -68,17 +72,11 @@ type publisher interface { Publish(ctx context.Context, subject string, payload []byte, opts ...jetstream.PublishOpt) (*jetstream.PubAck, error) } -// handle turns one event into its message on FORGE and acknowledges it. An event it cannot read is terminated, never -// 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) +// handle turns one pull request event into its message on FORGE and acknowledges it (bridge.Convert). An event it cannot +// read is terminated, never 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, files bridge.Files, watched bridge.Watching, logger *log.Logger) { + msg, err := bridge.Convert(m.Subject(), m.Data(), files, watched) switch { case errors.Is(err, bridge.ErrRetry): logger.Printf("%v; heard again in %s", err, retryDelay) @@ -98,18 +96,10 @@ func handleGrouped(ctx context.Context, m delivered, pub publisher, trees bridge if id := m.Headers().Get("x-event-id"); id != "" { opts = append(opts, jetstream.WithMsgID(id)) } - var member forgemsg.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) } @@ -147,7 +137,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro 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) } + // The branches the bridge watches (design 47 G §4, §507, §532): the registrations' first view and then + // registration-changed; before the first registration, the gate's source's branches and its source-moved. Nothing is + // written before both first views are held (ready). + watched := &bridge.Watched{} + said := func(ctx context.Context) error { return syncOpen(ctx, call, rec, heads, watched, &turn, logger) } // The module's one pull of its durable consumer on the shared events stream is its event hub (below). hub := process.NewEventHub(logger.Printf) // The state performed is a `requires` entry: opened by its binding, asked of the seat mesh-broker's show at this start @@ -176,9 +170,91 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } }() } + keptCall := keptCallOf(ctx, call) + reader := &views.ProcessReader{Module: forgeBridgeModule, Call: keptCall, Logf: logger.Printf, Hub: hub} + src := sourcebranches.New(sourcebranches.Call(keptCall)) + var regs *meshcatalogclient.Registrations + var mv *mover + recompute := func() { + if regs != nil { + watched.Set(registeredOf(regs.All()), unescaped(src.Branches())) + } + } + started := make(chan struct{}) // closed once the watched branches are read and the open pull requests said + regs, _, err = meshcatalogclient.WireRegistrationsProcess(reader, keptCall, func(mcseat.Registration, bool) { + recompute() + select { + case <-started: // a branch registered while the bridge runs is adopted: its head read (§532) + go mv.readHeads(ctx) + default: + } + }) + if err != nil { + return err + } + src.Follower.Changed = func(string, json.RawMessage, bool, time.Time) { recompute() } + if _, err := reader.Adopt(src.Follower, sourcebranches.Seat, sourcebranches.EventSourceMoved, src.Hear); err != nil { + return err + } + reader.Start(ctx, js, cred, nil) + // What a merge read of each module directory, by its tree (issue 1126): a merge reads only the directories it changed. + inputs := newKept(process.OptionalBucket(ctx, js, inputsBucket, "a merge's module directories are kept in memory only, so the first merge after a start reads every one", logger.Printf), logger) + mv = &mover{call: call, pub: rec, heads: heads, watched: watched, inputs: inputs, groups: groups, turn: &turn, logger: logger} + // The hub's handler of the forge's events, for every event its consumes filter lets through. Essential: the forge's + // events are what the bridge carries. The forge's events are the seat git's (novox/hq design 47 *The seat `git`*), + // whichever module holds it. + if err := hub.Handle("mesh.seat."+git.Seat+".event.>", func(m jetstream.Msg) { + select { + case <-started: + case <-ctx.Done(): + return + } + 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, filesVia(hctx, call), watched, logger) + }) + }); err != nil { + return err + } + if err := hub.Start(ctx, js, cred); err != nil { + return fmt.Errorf("%w: the bridge hears no forge event without it, so it stops; where the bus refused it or did not answer, this module's consumes (made by the controller with its grant) have not reached the bus yet", err) + } + defer hub.Stop() + // Every move of a branch, heard on the seat git's stream branch-moved through this module's durable there (§11), so a + // move said while the bridge was stopped is heard at its start. Essential: every merge and push reaches the stream so. + moves, err := cred.HearSeatEvents(ctx, js, git.StreamBranchMoved, func(m jetstream.Msg) { + select { + case <-started: + case <-ctx.Done(): + return + } + hctx, cancel := context.WithTimeout(ctx, syncDeadline) + defer cancel() + mv.hear(hctx, m) + }, logger.Printf) + if err != nil { + return fmt.Errorf("the seat git's branch-moved cannot be heard, so no merge or push reaches the forge stream: %w", err) + } + defer moves.Stop() + select { + case <-regs.Ready(): + case <-ctx.Done(): + return nil + } + if len(regs.All()) == 0 { + select { + case <-src.Follower.Ready(): + case <-ctx.Done(): + return nil + } + } + recompute() if err := said(ctx); err != nil { return err } + close(started) // What the gate and the checker say they owe, each on its seat's own stream owed through this module's durable there // (made from its consumes). A stream not there yet is said, and the bridge goes on: its other work does not wait on the // gate, and what was said meanwhile is caught up from the groups at the next start. @@ -217,25 +293,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro logger.Printf("%d acts owed to the forge, said while the bridge was away, performed from the groups", n) } go onReconnect(ctx, again, said, logger) - // What a merge read of each module directory, by its tree (issue 1126): a merge reads only the directories it changed. - inputs := newKept(process.OptionalBucket(ctx, js, inputsBucket, "a merge's module directories are kept in memory only, so the first merge after a start reads every one", logger.Printf), logger) - // The hub's handler of the forge's events, for every event its consumes filter lets through. Essential: the forge's - // events are what the bridge carries. - // The forge's events are the seat git's (novox/hq design 47 *The seat `git`*), whichever module holds it. - if err := hub.Handle("mesh.seat."+git.Seat+".event.>", 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() - handleGrouped(hctx, m, rec, treesVia(hctx, call), filesVia(hctx, call), inputs, groups, logger) - }) - }); err != nil { - return err - } - if err := hub.Start(ctx, js, cred); err != nil { - return fmt.Errorf("%w: the bridge hears no forge event without it, so it stops; where the bus refused it or did not answer, this module's consumes (made by the controller with its grant) have not reached the bus yet", err) - } - defer hub.Stop() + // Each watched branch's head, read at the start (§11): a head missed while the bridge was stopped is written now. + mv.readHeads(ctx) <-ctx.Done() return nil } diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go index da888759..b4874634 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go @@ -79,7 +79,7 @@ func TestReadableEventIsWrittenOnForgeAndAcked(t *testing.T) { // 1235/AC2 (unit }) var logs bytes.Buffer m, pub := &fakeMsg{subject: git.SubjectPullUpdated, data: body(nil)}, &fakePub{} - handle(context.Background(), m, pub, nil, someFiles, nil, log.New(&logs, "", 0)) + handle(context.Background(), m, pub, someFiles, watchesMain(), log.New(&logs, "", 0)) if len(pub.subjects) != 1 || pub.subjects[0] != "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-delivery.pull" { t.Fatalf("published on %v", pub.subjects) } @@ -95,7 +95,7 @@ func TestReadableEventIsWrittenOnForgeAndAcked(t *testing.T) { // 1235/AC2 (unit func TestUnreadableEventIsTerminatedAndLogged(t *testing.T) { // 1235/AC5 var logs bytes.Buffer m, pub := &fakeMsg{subject: git.SubjectPullUpdated, data: body(map[string]any{"head_sha": "short"})}, &fakePub{} - handle(context.Background(), m, pub, nil, someFiles, nil, log.New(&logs, "", 0)) + handle(context.Background(), m, pub, someFiles, watchesMain(), log.New(&logs, "", 0)) if len(pub.subjects) != 0 { t.Errorf("a message was written: %v", pub.subjects) } @@ -109,7 +109,7 @@ func TestUnreadableEventIsTerminatedAndLogged(t *testing.T) { // 1235/AC5 // from a test of its own, for the test budget. logs.Reset() m, pub = &fakeMsg{subject: git.SubjectPullUpdated, data: body(nil)}, &fakePub{err: errors.New("no stream")} - handle(context.Background(), m, pub, nil, someFiles, nil, log.New(&logs, "", 0)) + handle(context.Background(), m, pub, someFiles, watchesMain(), log.New(&logs, "", 0)) if m.acked != 0 || m.termd != 0 { t.Errorf("acked %d terminated %d, want neither", m.acked, m.termd) } @@ -118,55 +118,93 @@ func TestUnreadableEventIsTerminatedAndLogged(t *testing.T) { // 1235/AC5 } } -// mergedEvent is the seat git's pull.merged. -func mergedEvent() []byte { - b, _ := json.Marshal(map[string]any{"repository": "novox/mesh-delivery", "number": 12, "branch": "feat/x", "head": head, - "base": "main", "merge-commit": "89abcdef0123456789abcdef0123456789abcdef"}) +// watchesMain watches main of every repository, as the gate's source names it before the first registration. +func watchesMain() *bridge.Watched { + w := &bridge.Watched{} + w.Set(nil, []string{"main"}) + return w +} + +// movedEvent is the seat git's branch-moved of main of novox/mesh-delivery, naming a head the bridge never takes. +func movedEvent(branch string) []byte { + b, _ := json.Marshal(map[string]any{"repository": "novox/mesh-delivery", "branch": branch, "head": strings.Repeat("f", 40)}) return b } +const landed = "89abcdef0123456789abcdef0123456789abcdef" + +// moveCall answers the seat git as a forge whose main of novox/mesh-delivery is protected at landed, merged there by #12; +// trees answers trees, every other verb answers err. +func moveCall(trees json.RawMessage, err error, keys *[]string, args *[]any, mu *sync.Mutex) caller { + return func(_ context.Context, key string, body any) (json.RawMessage, error) { + mu.Lock() + *keys, *args = append(*keys, key), append(*args, body) + mu.Unlock() + switch key { + case "seat:git.branch": + return json.RawMessage(`{"repository":"novox/mesh-delivery","branch":"main","head":"` + landed + `","protected":true}`), nil + case "seat:git.pull": + return json.RawMessage(`{"pull":{"repository":"novox/mesh-delivery","number":12,"branch":"feat/x","head":"` + head + `","base":"main","state":"merged","merge-commit":"` + landed + `"}}`), nil + case "seat:git.trees": + if trees != nil { + return trees, nil + } + } + return nil, err + } +} + +func newMover(call caller, pub publisher, heads headStore, logs *bytes.Buffer) *mover { + return &mover{call: call, pub: pub, heads: heads, watched: watchesMain(), turn: &sync.Mutex{}, logger: log.New(logs, "", 0)} +} + func TestTreesRefusedLeavesTheEventForTenSecondsAndLogs(t *testing.T) { // 1236/AC3 var logs bytes.Buffer - m, pub := &fakeMsg{subject: git.SubjectPullMerged, data: mergedEvent()}, &fakePub{} - call := func(context.Context, string, any) (json.RawMessage, error) { - return nil, errors.New("the tree listing of novox/mesh-delivery was cut") - } - handle(context.Background(), m, pub, treesVia(context.Background(), call), filesVia(context.Background(), call), nil, log.New(&logs, "", 0)) - if len(pub.subjects) != 0 { + var keys []string + var args []any + var mu sync.Mutex + m, pub, heads := &fakeMsg{subject: bridge.BranchMovedSubject, data: movedEvent("main")}, &fakePub{}, newHeads() + newMover(moveCall(nil, errors.New("the tree listing of novox/mesh-delivery was cut"), &keys, &args, &mu), pub, heads, &logs).hear(context.Background(), m) + if len(pub.subjects) != 0 || heads.get("branch.novox.mesh-delivery.main") != "" { t.Errorf("a message was written: %v", pub.subjects) } if m.acked != 0 || m.termd != 0 || len(m.nak) != 1 || m.nak[0] < 10*time.Second { t.Errorf("acked %d terminated %d naks %v, want one nak of at least 10 s", m.acked, m.termd, m.nak) } - for _, w := range []string{"novox/mesh-delivery", "89abcdef0123456789abcdef0123456789abcdef", "the tree listing of novox/mesh-delivery was cut"} { + for _, w := range []string{"novox/mesh-delivery", landed, "the tree listing of novox/mesh-delivery was cut"} { if !strings.Contains(logs.String(), w) { t.Errorf("log %q does not name %q", logs.String(), w) } } } -func TestMergeCallsTheToolOnceAndWritesTheMergedMessage(t *testing.T) { // 1236/AC1; design 47 *The seat git* (verbs trees, file) +// 1236/AC1; design 47 *The seat git* (verbs trees, file); design 47 G §11, §20, §509 (FB1, FB10): a move of a watched branch +// is taken by the seat's verb branch, never the event's head, written once as merged with its pull request and base, and +// its head kept, so the same move heard again writes nothing; a move of a branch not watched writes nothing. +func TestMergeCallsTheToolOnceAndWritesTheMergedMessage(t *testing.T) { var logs bytes.Buffer - m, pub := &fakeMsg{subject: git.SubjectPullMerged, data: mergedEvent()}, &fakePub{} var keys []string var args []any var mu sync.Mutex // the files are asked for concurrently (issue 1458) trees := gitfixture.Given(t, "trees answers", "verb:trees.answer") // root ddd…, "" ddd…, contracts eee… + call := moveCall(trees, nil, &keys, &args, &mu) file := gitfixture.Given(t, "file answers", "verb:file.answer") - call := func(_ context.Context, key string, body any) (json.RawMessage, error) { - mu.Lock() - keys, args = append(keys, key), append(args, body) - mu.Unlock() + files := func(ctx context.Context, key string, body any) (json.RawMessage, error) { if key == "seat:git.file" { // the manifests, 1448; the trees are what this test is about + mu.Lock() + keys = append(keys, key) + mu.Unlock() return file, nil } - return trees, nil + return call(ctx, key, body) } - handle(context.Background(), m, pub, treesVia(context.Background(), call), filesVia(context.Background(), call), nil, log.New(&logs, "", 0)) - if len(keys) == 0 || keys[0] != "seat:git.trees" || slices.Index(keys[1:], "seat:git.trees") >= 0 { - t.Fatalf("calls %v, want the trees once and first", keys) + m, pub, heads := &fakeMsg{subject: bridge.BranchMovedSubject, data: movedEvent("main")}, &fakePub{}, newHeads() + mv := newMover(files, pub, heads, &logs) + mv.hear(context.Background(), m) + if len(keys) < 3 || keys[0] != "seat:git.branch" || keys[1] != "seat:git.pull" || keys[2] != "seat:git.trees" || slices.Index(keys[3:], "seat:git.trees") >= 0 { + t.Fatalf("calls %v, want the branch, its pull, then the trees once", keys) } - if got, _ := json.Marshal(args[0]); !sameJSON(got, []byte(`{"repository":"novox/mesh-delivery","commit":"89abcdef0123456789abcdef0123456789abcdef"}`)) { + if got, _ := json.Marshal(args[2]); !sameJSON(got, []byte(`{"repository":"novox/mesh-delivery","commit":"`+landed+`"}`)) { t.Errorf("arguments %s", got) } if len(pub.subjects) != 1 || pub.subjects[0] != "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-delivery.merged" { @@ -174,11 +212,17 @@ func TestMergeCallsTheToolOnceAndWritesTheMergedMessage(t *testing.T) { // 1236/ } f, err := forgemsg.DecodeForge(pub.data[0]) d, e := strings.Repeat("d", 40), strings.Repeat("e", 40) - if err != nil || f.Root != d || f.Trees[""] != d || f.Trees["contracts"] != e || f.Merge == "" { + if err != nil || f.Root != d || f.Trees[""] != d || f.Trees["contracts"] != e || f.Merge != landed || f.Base != "main" || f.Number != 12 || f.Branch != "feat/x" { t.Errorf("message %+v err %v", f, err) } - if m.acked != 1 { - t.Errorf("acked %d", m.acked) + if m.acked != 1 || heads.get("branch.novox.mesh-delivery.main") != landed { + t.Errorf("acked %d, head kept %q", m.acked, heads.get("branch.novox.mesh-delivery.main")) + } + again, other := &fakeMsg{subject: bridge.BranchMovedSubject, data: movedEvent("main")}, &fakeMsg{subject: bridge.BranchMovedSubject, data: movedEvent("feat/x")} + mv.hear(context.Background(), again) + mv.hear(context.Background(), other) + if len(pub.subjects) != 1 || again.acked != 1 || other.acked != 1 { + t.Errorf("the same move again and a move of a branch not watched wrote %d messages (acked %d, %d), want none", len(pub.subjects)-1, again.acked, other.acked) } } diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/moves.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/moves.go new file mode 100644 index 00000000..40b6c060 --- /dev/null +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/moves.go @@ -0,0 +1,143 @@ +package main + +import ( + "context" + "errors" + "fmt" + "log" + "strings" + "sync" + + "github.com/nats-io/nats.go/jetstream" + + mcseat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" + "git.novox.be/novox/mesh-sdk/go/keys" + + "mesh-delivery-forge-bridge/internal/bridge" +) + +// mover writes the one message of each move of a watched branch (design 47 G §11, §509, §714): heard on the seat git's +// branch-moved through the bridge's durable on SEAT_git_branch-moved, the head read by the seat's verb branch, never the +// event's; merged with its pull request, or pushed; the head written kept in the state heads under the branch's key +// (A-236), so a move heard twice writes once and the next pushed says its head before. +type mover struct { + call caller + pub publisher + heads headStore + watched *bridge.Watched + inputs bridge.Kept + groups *grouper + turn *sync.Mutex + logger *log.Logger +} + +// hear takes one branch-moved: one it cannot read is terminated, one of a branch not watched is acknowledged with nothing +// written, one the forge could not say about is heard again. +func (mv *mover) hear(ctx context.Context, m delivered) { + moved, err := bridge.MovedOf(m.Data()) + if err != nil { + mv.logger.Printf("%v; the event is dropped", err) + _ = m.Term() + return + } + if !mv.watched.Watches(moved.Repository(), moved.Branch) { + _ = m.Ack() + return + } + mv.turn.Lock() + err = mv.move(ctx, moved) + mv.turn.Unlock() + switch { + case errors.Is(err, bridge.ErrUnreadable): + mv.logger.Printf("%v; the event is dropped", err) + _ = m.Term() + case err != nil: + mv.logger.Printf("%v; heard again in %s", err, retryDelay) + _ = m.NakWithDelay(retryDelay) + default: + _ = m.Ack() + } +} + +// move reads the branch's head and writes its message, unless that head is the one last written for it. +func (mv *mover) move(ctx context.Context, moved bridge.Moved) error { + b, err := bridge.ReadBranch(ctx, bridge.Caller(mv.call), moved) + if err != nil { + return fmt.Errorf("%s of %s: %w", moved.Branch, moved.Repository(), err) + } + before, _, err := mv.heads.Get(ctx, moved.HeadKey()) + if err != nil { + return fmt.Errorf("%s of %s: the head last written cannot be read: %w", moved.Branch, moved.Repository(), err) + } + if before == b.Head { + return nil + } + if !b.Protected { + // A branch whose protection was removed starts nothing: nothing is written (§17, §509). + mv.logger.Printf("%s of %s moved to %s while it is not protected: nothing is written", moved.Branch, moved.Repository(), b.Head) + return nil + } + msg, err := bridge.MoveMessage(ctx, bridge.Caller(mv.call), moved, b.Head, before, treesVia(ctx, mv.call), filesVia(ctx, mv.call), mv.inputs) + if err != nil { + return err + } + data, grouped := msg.Data, false + var member forgemsg.Forge + if mv.groups != nil { + data, member, grouped = mv.groups.member(ctx, msg.Data) + } + // The branch's new head makes a write after a lost acknowledgement the same message to the stream. + id := "moved." + keys.EscapeName(moved.Repository()) + "." + keys.EscapeName(moved.Branch) + "." + b.Head + if _, err := mv.pub.Publish(ctx, forgeSubject(msg.Event), data, jetstream.WithMsgID(id)); err != nil { + return fmt.Errorf("%s could not be written on FORGE: %w", msg.Event, err) + } + if grouped { + mv.groups.published(ctx, member) + } + if err := mv.heads.Put(ctx, moved.HeadKey(), b.Head); err != nil { + mv.logger.Printf("%s of %s: the head %s written is not kept, so the next move says the older head before: %v", moved.Branch, moved.Repository(), b.Head, err) + } + if msg.Said != "" { + mv.logger.Printf("%s", msg.Said) + } + return nil +} + +// readHeads reads each watched branch's head (§11): a head missed while the bridge was stopped is written now. +func (mv *mover) readHeads(ctx context.Context) { + for _, repo := range mv.watched.Repositories() { + owner, name, ok := strings.Cut(repo, "/") + if !ok { + continue + } + for _, branch := range mv.watched.Branches(repo) { + mv.turn.Lock() + err := mv.move(ctx, bridge.Moved{Owner: owner, Repo: name, Branch: branch}) + mv.turn.Unlock() + if err != nil { + mv.logger.Printf("the head of %s of %s is not read at this start, so a move missed meanwhile waits for its next move: %v", branch, repo, err) + } + } + } +} + +// registeredOf is the registrations as the watched branches read them. +func registeredOf(regs []mcseat.Registration) []bridge.Registered { + out := make([]bridge.Registered, 0, len(regs)) + for _, r := range regs { + out = append(out, bridge.Registered{Origin: r.Origin, Branches: r.Branches}) + } + return out +} + +// unescaped is the gate's source's branches by name (they are kept by their escaped name). +func unescaped(branches []string) []string { + out := make([]string, 0, len(branches)) + for _, b := range branches { + if name, err := keys.UnescapeName(b); err == nil { + out = append(out, name) + } + } + return out +} 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 9441a818..826d5295 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 @@ -80,7 +80,7 @@ func hear(mu *sync.Mutex, react func()) { // 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 { +func syncOpen(ctx context.Context, call caller, pub publisher, heads headStore, watched bridge.Watching, mu *sync.Mutex, logger *log.Logger) error { mu.Lock() defer mu.Unlock() ctx, cancel := context.WithTimeout(ctx, syncDeadline) @@ -89,11 +89,11 @@ func syncOpen(ctx context.Context, call caller, pub publisher, heads headStore, if err != nil { return fmt.Errorf("%s: %w", openPullsTool, err) } - msgs, skipped, err := bridge.OpenPulls(raw) + msgs, skipped, err := bridge.OpenPulls(raw, watched) if err != nil { return fmt.Errorf("%s: %w", openPullsTool, err) } - open, err := bridge.OpenMessage(raw) + open, err := bridge.OpenMessage(raw, watched) if err != nil { return fmt.Errorf("%s: %w", openPullsTool, err) } 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 5861b9b7..a10a62f3 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 @@ -90,7 +90,7 @@ func (m *memHeads) get(key string) string { m.mu.Lock(); defer m.mu.Unlock(); re // 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) + return syncOpen(context.Background(), call, &recorder{pub: pub, heads: heads, logger: logger}, heads, watchesMain(), mu, logger) } // 1279/AC6: the sync writes a pull message only for a head that differs from what FORGE holds, then one open message diff --git a/modules/mesh-delivery-forge-bridge/forgemsg/forge.go b/modules/mesh-delivery-forge-bridge/forgemsg/forge.go index ccba891b..027a891c 100644 --- a/modules/mesh-delivery-forge-bridge/forgemsg/forge.go +++ b/modules/mesh-delivery-forge-bridge/forgemsg/forge.go @@ -24,8 +24,11 @@ const ForgeStream = "forge" // The kinds of forge message, the last part of a forge event's name. const ( - ForgePull = "pull" - ForgeMerged = "merged" + ForgePull = "pull" + ForgeMerged = "merged" + // ForgePushed is a move of a watched branch the forge names no pull request's merge for (design 47 G §509): an + // admitted push, taken in as a merged message is. + ForgePushed = "pushed" ForgeRefused = "refused" ForgeOpen = "open" ) @@ -67,10 +70,10 @@ func checkRepository(repository string) error { func checkForgeKind(kind string) error { switch kind { - case ForgePull, ForgeMerged, ForgeRefused, ForgeOpen: + case ForgePull, ForgeMerged, ForgePushed, ForgeRefused, ForgeOpen: return nil } - return fmt.Errorf("delivery: the forge kind %q is not pull, merged, refused or open", kind) + return fmt.Errorf("delivery: the forge kind %q is not pull, merged, pushed, refused or open", kind) } // ForgeEvent is the name of a forge message of the given kind: `forge...`. @@ -111,18 +114,24 @@ func ParseForgeEvent(event string) (repository, kind string, err error) { // DirsKnown, true when the forge could say them: known and empty is DirsKnown with no Dirs, not known is neither // (the forge's module_dirs and module_dirs_said, mesh-catalog modules/gitea/trunk.ts). Other messages carry neither. type Forge struct { - V int `json:"v"` - Kind string `json:"kind"` - Repository string `json:"repository"` - Branch string `json:"branch,omitempty"` - Number int `json:"number,omitempty"` - Head string `json:"head,omitempty"` - Action string `json:"action,omitempty"` - Dirs []string `json:"dirs,omitempty"` - DirsKnown bool `json:"dirs-known,omitempty"` - Merge string `json:"merge,omitempty"` - Trees map[string]string `json:"trees,omitempty"` - Root string `json:"root,omitempty"` + V int `json:"v"` + Kind string `json:"kind"` + Repository string `json:"repository"` + // Branch is a pull or merged message's pull request branch, and a pushed message's branch pushed. + Branch string `json:"branch,omitempty"` + // Base is the watched branch a merged message's commit landed on (design 47 G §714). + Base string `json:"base,omitempty"` + Number int `json:"number,omitempty"` + // Head is a pull or merged message's pull request head commit, and a pushed message's new head of its branch (A-236). + Head string `json:"head,omitempty"` + // Before is a pushed message's head of its branch before, as the forge bridge last wrote it (A-236), empty for none. + Before string `json:"before,omitempty"` + Action string `json:"action,omitempty"` + Dirs []string `json:"dirs,omitempty"` + DirsKnown bool `json:"dirs-known,omitempty"` + Merge string `json:"merge,omitempty"` + Trees map[string]string `json:"trees,omitempty"` + Root string `json:"root,omitempty"` // Manifests are the module.json bytes, exact, of each module directory: of the merge's trees on a merged message, // of Dirs at the head on a pull message. Locks are, of each such directory, its lock files' hex sha256 by path // relative to it (novox/hq ADR 0465 clause 1). Older writers send neither. @@ -203,8 +212,8 @@ func (f Forge) checkPredefined() error { if len(f.Predefined) == 0 && !f.PredefinedRead { return nil } - if f.Kind != ForgeMerged { - return fmt.Errorf("delivery: only a merged message carries the predefined list, not a %s message", f.Kind) + if f.Kind != ForgeMerged && f.Kind != ForgePushed { + return fmt.Errorf("delivery: only a merged or pushed message carries the predefined list, not a %s message", f.Kind) } if len(f.Predefined) > 0 && f.PredefinedRead { return fmt.Errorf("delivery: a merged message carries the predefined list and says the merge holds none (predefined-read)") @@ -262,21 +271,27 @@ func (f Forge) check() error { if err := f.checkInputs(func(d string) bool { return dirs[d] }, "its module directories"); err != nil { return err } - case ForgeMerged: + case ForgeMerged, ForgePushed: if len(f.Dirs) > 0 || f.DirsKnown { - return fmt.Errorf("delivery: a merged message carries no module directories") + return fmt.Errorf("delivery: a %s message carries no module directories", f.Kind) } - if f.Number <= 0 { + if f.Kind == ForgeMerged && f.Number <= 0 { return fmt.Errorf("delivery: a merged message needs its pull request number") } - if f.Head == "" { - return fmt.Errorf("delivery: a merged message needs its head") + if f.Kind == ForgePushed && (f.Number != 0 || f.Merge != "" || f.Base != "") { + return fmt.Errorf("delivery: a pushed message names no pull request, merge commit or base: its branch is the one pushed") } - if f.Merge == "" { + if f.Kind == ForgePushed && f.Branch == "" { + return fmt.Errorf("delivery: a pushed message needs its branch") + } + if f.Head == "" { + return fmt.Errorf("delivery: a %s message needs its head", f.Kind) + } + if f.Kind == ForgeMerged && f.Merge == "" { return fmt.Errorf("delivery: a merged message needs its merge commit") } if f.Root == "" { - return fmt.Errorf("delivery: a merged message needs the repository's root tree") + return fmt.Errorf("delivery: a %s message needs the repository's root tree", f.Kind) } // No trees is a merge that leaves the repository with no module.json anywhere (a documents repository, or one // whose last module was removed): it is written all the same, so the indexer takes the merge in and drops the diff --git a/modules/mesh-delivery-forge-bridge/forgemsg/forge_test.go b/modules/mesh-delivery-forge-bridge/forgemsg/forge_test.go index 41fde65b..c31695e5 100644 --- a/modules/mesh-delivery-forge-bridge/forgemsg/forge_test.go +++ b/modules/mesh-delivery-forge-bridge/forgemsg/forge_test.go @@ -23,6 +23,10 @@ func sampleForges() []Forge { // 1585/T1: a group member's merged message names its group and every approved member (ADR 0633 §4). {Kind: ForgeMerged, Repository: "novox/mesh-sdk", Number: 37, Head: "dddd", Merge: "eeee", Root: "t0", Group: "feat/take-in", Members: []meshdeliveryclient.Head{{Repository: "novox/mesh-sdk", Number: 37, Head: "dddd"}, {Repository: "novox/mesh-delivery", Number: 9, Head: "ffff"}}}, + // design 47 G §509, §714, A-236: a merged message names the watched branch it landed on; a pushed one its branch, new + // head and head before. + {Kind: ForgeMerged, Repository: "novox/mesh-sdk", Number: 38, Branch: "feat/y", Head: "aaaa", Merge: "bbbb", Base: "beta", Root: "t0"}, + {Kind: ForgePushed, Repository: "novox/mesh-sdk", Branch: "beta", Head: "cccc", Before: "bbbb", Root: "t0", Trees: map[string]string{"modules/a": "t1"}}, } } @@ -153,7 +157,7 @@ func TestForge(t *testing.T) { {"assignments read", "mesh-module-assigner", in(func(f *Forge) { f.RetiredRead = true })}, {"predefined not an object", "one JSON object", in(func(f *Forge) { f.Predefined = []byte(`["domain"]`) })}, {"predefined both", "says the merge holds none", in(func(f *Forge) { f.Predefined = []byte(`{}`); f.PredefinedRead = true })}, - {"pull with the predefined list", "only a merged message", Forge{Kind: ForgePull, Repository: "novox/mesh-sdk", Number: 1, Head: "h", Action: PullOpened, PredefinedRead: true}}, + {"pull with the predefined list", "only a merged or pushed message", Forge{Kind: ForgePull, Repository: "novox/mesh-sdk", Number: 1, Head: "h", Action: PullOpened, PredefinedRead: true}}, {"pull manifest outside dirs", `"b"`, pullIn(func(f *Forge) { f.Manifests["b"] = "{}" })}, {"pull locks outside dirs", `"c"`, pullIn(func(f *Forge) { f.Locks["c"] = map[string]string{"go.sum": "h"} })}, {"refused with manifests", "no manifests", Forge{Kind: ForgeRefused, Repository: "a/b", Act: "x", Words: "no", Manifests: map[string]string{"a": "{}"}}}, @@ -167,6 +171,9 @@ func TestForge(t *testing.T) { {"merged with members leaving itself out", "leave out", at(4, func(f *Forge) { f.Members = f.Members[1:] })}, {"merged with a member twice", "twice", at(4, func(f *Forge) { f.Members = append(f.Members, f.Members[0]) })}, {"open with a group", "only the open pull requests", Forge{Kind: ForgeOpen, Group: "g"}}, + // design 47 G §509, A-236: a pushed message names its branch and no pull request, merge commit or base. + {"pushed with a pull request", "names no pull request", Forge{Kind: ForgePushed, Repository: "a/b", Branch: "beta", Head: "h", Root: "r", Number: 3}}, + {"pushed with no branch", "needs its branch", Forge{Kind: ForgePushed, Repository: "a/b", Head: "h", Root: "r"}}, } { if _, err := EncodeForge(c.f); err == nil || !strings.Contains(err.Error(), c.want) { t.Errorf("%s: %v, want a refusal naming %q", c.name, err, c.want) diff --git a/modules/mesh-delivery-forge-bridge/go.mod b/modules/mesh-delivery-forge-bridge/go.mod index e8f3740d..ea862348 100644 --- a/modules/mesh-delivery-forge-bridge/go.mod +++ b/modules/mesh-delivery-forge-bridge/go.mod @@ -4,7 +4,9 @@ go 1.26.0 require ( git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshdeliveryclient v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/git v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdeliverychecker v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery v0.0.0-00010101000000-000000000000 diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go index d27bb022..21648ba3 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go @@ -32,8 +32,10 @@ var ErrNotHandled = errors.New("event not handled") // heard again later. var ErrRetry = errors.New("the forge could not say") -// Trunk is the branch a merged message is written for. -const Trunk = "main" +// Watching says whether the bridge watches a branch of a repository (Watched; design 47 G §507, §532). +type Watching interface { + Watches(repository, branch string) bool +} // Trees asks the forge for the trees of a commit: the root tree and the tree of every directory holding a module.json // (the root directory spelled ""), as the seat git's verb trees answers (issue 1234/AC4). @@ -93,7 +95,7 @@ type BranchDeleted struct { } // branchDeleted turns forge.delete into a branch-deleted message; a deleted tag is not handled. -func branchDeleted(subject string, data []byte) (Message, error) { +func branchDeleted(subject string, data []byte, watched Watching) (Message, error) { var e git.ForgeDeleteEvent if err := json.Unmarshal(data, &e); err != nil { return Message{}, unreadable(subject, "the body is not json: %v", err) @@ -109,8 +111,8 @@ func branchDeleted(subject string, data []byte) (Message, error) { return Message{}, unreadable(subject, "no owner or repo a forge event can name (%q/%q)", d.Owner, d.Repo) case branch == "": return Message{}, unreadable(subject, "no branch") - case branch == Trunk: - return Message{}, fmt.Errorf("%w: %s of the trunk %s", ErrNotHandled, subject, Trunk) + case watched.Watches(d.Owner+"/"+d.Repo, branch): + return Message{}, fmt.Errorf("%w: %s of the watched branch %s", ErrNotHandled, subject, branch) } body, err := json.Marshal(BranchDeleted{V: keys.Version, Kind: ForgeBranchDeleted, Repository: d.Owner + "/" + d.Repo, Branch: branch}) if err != nil { @@ -170,29 +172,20 @@ func unreadable(subject, why string, args ...any) error { return fmt.Errorf("%w: %s: %s", ErrUnreadable, subject, fmt.Sprintf(why, args...)) } -// Convert turns one event of the forge module, by its bus subject and body, into a message. An event it cannot read -// is an ErrUnreadable error naming the subject and why; an event it writes nothing for is ErrNotHandled. A merge into -// the trunk asks trees once and becomes a merged message; a merge into any other branch is a closed pull message, as the -// pull request left its group. A forge that cannot say is ErrRetry. A merge that leaves no module.json anywhere is a merged message with no -// trees all the same (issue 1126): the indexer takes it in and drops the repository's modules, and the view reads the -// repository as reached at it. -func Convert(subject string, data []byte, trees Trees, files Files) (Message, error) { - return ConvertKept(subject, data, trees, files, nil) -} - -// ConvertKept is Convert; with kept, a merge reads the forge only for the module directories whose tree hash kept -// does not hold, and keeps what it read (kept.go). -func ConvertKept(subject string, data []byte, trees Trees, files Files, kept Kept) (Message, error) { +// Convert turns one pull request event of the seat git, by its bus subject and body, into a message. An event it cannot +// read is an ErrUnreadable error naming the subject and why; an event it writes nothing for is ErrNotHandled. A pull +// request into a branch the bridge does not watch is no candidate: written closed. A merge into a watched branch writes +// nothing here: its merged message is the move's, heard on branch-moved (MoveMessage; design 47 G §509, §714); a merge +// into another branch is a closed pull message, as the pull request left its group. A forge that cannot say is ErrRetry. +func Convert(subject string, data []byte, files Files, watched Watching) (Message, error) { var action string switch subject { case UpdatedSubject: action = forgemsg.PullUpdated - case ClosedSubject: + case ClosedSubject, MergedSubject: action = forgemsg.PullClosed - case MergedSubject: - action = forgemsg.PullClosed // unless it merged into the trunk, below case DeletedSubject: - return branchDeleted(subject, data) + return branchDeleted(subject, data, watched) default: return Message{}, fmt.Errorf("%w: %s", ErrNotHandled, subject) } @@ -210,17 +203,17 @@ func ConvertKept(subject string, data []byte, trees Trees, files Files, kept Kep case p.Head == "": return Message{}, unreadable(subject, "no branch") } - if subject == MergedSubject && p.Base == Trunk { - return merged(subject, p, trees, files, kept) + into := p.Base != "" && watched.Watches(p.Owner+"/"+p.Repo, p.Base) + if subject == MergedSubject && into { + return Message{}, fmt.Errorf("%w: %s of %s/%s#%d into the watched branch %s: its merged message is written from the branch's move", ErrNotHandled, subject, p.Owner, p.Repo, p.Number, p.Base) } said := "" - if action == forgemsg.PullUpdated && p.Base != "" && p.Base != Trunk { - // The checker judges pull requests into main only, and source.main names main's head only (novox/hq ADR 0484 §11): a - // pull request into another base is no candidate, as it is none at the start (OpenPulls) and at its merge. It is - // written closed, so a candidate an older head made is taken out, and said (issue 1422). + if action == forgemsg.PullUpdated && p.Base != "" && !into { + // A pull request into a branch no registration watches is no candidate, as it is none at the start (OpenPulls) and at + // its merge. It is written closed, so a candidate an older head made is taken out, and said (issue 1422). action = forgemsg.PullClosed - said = fmt.Sprintf("%s/%s#%d at %s is into %s, not %s: written closed, so it is no candidate; the checker judges pull requests into %s only (novox/hq ADR 0484 §11)", - p.Owner, p.Repo, p.Number, p.HeadSHA[:12], p.Base, Trunk, Trunk) + said = fmt.Sprintf("%s/%s#%d at %s is into %s, which is not watched: written closed, so it is no candidate", + p.Owner, p.Repo, p.Number, p.HeadSHA[:12], p.Base) } f := forgemsg.Forge{Kind: forgemsg.ForgePull, Repository: p.Owner + "/" + p.Repo, Branch: p.Head, Number: p.Number, Head: p.HeadSHA, Action: action} @@ -419,15 +412,17 @@ func fetch(owner, repo, ref string, reads []*read, files Files) { wg.Wait() } -// merged is the merged message of a merge into the trunk: the trees of the merge commit, asked once. -func merged(subject string, p pull, trees Trees, files Files, kept Kept) (Message, error) { - if !sha40.MatchString(p.MergeSHA) { - return Message{}, unreadable(subject, "the merge commit %q is not 40 hex digits", p.MergeSHA) +// landed is the message of a commit that landed on a watched branch, merged or pushed (f says which and what names it): +// the trees of the commit, asked once, and the take-in inputs of every module directory at it (design 47 G §509, A-236). +func landed(what string, f forgemsg.Forge, commit string, trees Trees, files Files, kept Kept) (Message, error) { + if !sha40.MatchString(commit) { + return Message{}, unreadable(what, "the commit %q is not 40 hex digits", commit) } - repository := p.Owner + "/" + p.Repo - root, dirs, err := trees(p.Owner, p.Repo, p.MergeSHA) + repository := f.Repository + owner, repo, _ := strings.Cut(repository, "/") + root, dirs, err := trees(owner, repo, commit) if err != nil { - return Message{}, fmt.Errorf("%w: %s: the trees of %s at the merge commit %s: %v", ErrRetry, subject, repository, p.MergeSHA, err) + return Message{}, fmt.Errorf("%w: %s: the trees of %s at %s: %v", ErrRetry, what, repository, commit, err) } names := make([]string, 0, len(dirs)) for dir := range dirs { @@ -435,10 +430,10 @@ func merged(subject string, p pull, trees Trees, files Files, kept Kept) (Messag } slices.Sort(names) manifests, locks, needs, toRead := keptInputs(names, dirs, kept) - read, readLocks, readNeeds, whole, err := inputsRead(p.Owner, p.Repo, p.MergeSHA, toRead, files) + read, readLocks, readNeeds, whole, err := inputsRead(owner, repo, commit, toRead, files) if err != nil { keep(whole, dirs, read, readLocks, readNeeds, kept) // what was read whole is not asked again at the next try - return Message{}, fmt.Errorf("%w: %s: the manifests of %s at the merge commit %s (%d of %d module directories read from the forge, the others kept by their tree): %v", ErrRetry, subject, repository, p.MergeSHA, len(toRead), len(names), err) + return Message{}, fmt.Errorf("%w: %s: the manifests of %s at %s (%d of %d module directories read from the forge, the others kept by their tree): %v", ErrRetry, what, repository, commit, len(toRead), len(names), err) } keep(toRead, dirs, read, readLocks, readNeeds, kept) maps.Copy(manifests, read) @@ -453,31 +448,30 @@ func merged(subject string, p pull, trees Trees, files Files, kept Kept) (Messag if len(needs) == 0 { needs = nil } - event, err := forgemsg.ForgeEvent(repository, forgemsg.ForgeMerged) + event, err := forgemsg.ForgeEvent(repository, f.Kind) if err != nil { - return Message{}, unreadable(subject, "%v", err) + return Message{}, unreadable(what, "%v", err) } - f := forgemsg.Forge{Kind: forgemsg.ForgeMerged, Repository: repository, Branch: p.Head, Number: p.Number, - Head: p.HeadSHA, Merge: p.MergeSHA, Root: root, Trees: dirs, Manifests: manifests, Locks: locks, Needs: needs} + f.Root, f.Trees, f.Manifests, f.Locks, f.Needs = root, dirs, manifests, locks, needs if repository == CatalogRepository { - // The catalogue's predefined list at the merge (novox/hq ADR 0565 §2): the indexer keeps it for mesh-config. - raw, err := files(p.Owner, p.Repo, PredefinedFile, p.MergeSHA) + // The catalogue's predefined list at the commit (novox/hq ADR 0565 §2): the indexer keeps it for mesh-config. + raw, err := files(owner, repo, PredefinedFile, commit) switch { case errors.Is(err, ErrNotFound): f.PredefinedRead = true case err != nil: - return Message{}, fmt.Errorf("%w: %s: %s of %s at the merge commit %s: %v", ErrRetry, subject, PredefinedFile, repository, p.MergeSHA, err) + return Message{}, fmt.Errorf("%w: %s: %s of %s at %s: %v", ErrRetry, what, PredefinedFile, repository, commit, err) default: f.Predefined = json.RawMessage(raw) } } body, err := forgemsg.EncodeForge(f) if err != nil { - return Message{}, unreadable(subject, "%v", err) + return Message{}, unreadable(what, "%v", err) } msg := Message{Event: event, Data: body} if len(dirs) == 0 { - msg.Said = fmt.Sprintf("%s at the merge commit %s holds no module.json: a merged message with no module directories", repository, p.MergeSHA) + msg.Said = fmt.Sprintf("%s at %s holds no module.json: a %s message with no module directories", repository, commit, f.Kind) } return msg, nil } @@ -506,16 +500,16 @@ func openPulls(answer []byte) ([]pull, error) { } // OpenPulls turns the seat git's answer of open-pulls into one pull message per open pull request into -// the trunk, action updated and its directories not known (the forge's answer does not say them). A pull request into +// a watched branch, action updated and its directories not known (the forge's answer does not say them). A pull request into // another branch gets none. An entry that cannot be read is named in skipped and gets none; an answer that is not // the contract's, or that says an error, is an error. -func OpenPulls(answer []byte) (msgs []Message, skipped []string, err error) { +func OpenPulls(answer []byte, watched Watching) (msgs []Message, skipped []string, err error) { pulls, err := openPulls(answer) if err != nil { return nil, nil, err } for _, p := range pulls { - if p.Base != Trunk { + if !watched.Watches(p.Owner+"/"+p.Repo, p.Base) { continue } name := fmt.Sprintf("%s/%s#%d", p.Owner, p.Repo, p.Number) @@ -553,17 +547,17 @@ func Track(f forgemsg.Forge) (key, head string, ok bool) { return "", "", false } -// OpenMessage turns the seat git's answer of open-pulls into the open message: every open pull request into the trunk, by +// OpenMessage turns the seat git's answer of open-pulls into the open message: every open pull request into a watched branch, 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) { +func OpenMessage(answer []byte, watched Watching) (Message, error) { pulls, err := openPulls(answer) if err != nil { return Message{}, err } open := []forgemsg.OpenPull{} for _, p := range pulls { - if p.Base != Trunk || p.Owner == "" || p.Repo == "" || p.Number <= 0 { + if !watched.Watches(p.Owner+"/"+p.Repo, p.Base) || p.Owner == "" || p.Repo == "" || p.Number <= 0 { continue } open = append(open, forgemsg.OpenPull{Repository: p.Owner + "/" + p.Repo, Number: p.Number}) 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 5b6b2efb..dc0aaa35 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge_test.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge_test.go @@ -68,7 +68,7 @@ const updatedSubject = "mesh.seat.git.event.pull-updated" const closedSubject = "mesh.seat.git.event.pull-closed" func TestUpdatedBecomesAPullMessage(t *testing.T) { // 1235/AC2 (the conversion; the bus is the integration test) - msg, err := Convert(updatedSubject, updated(nil), nil, anyFiles) + msg, err := Convert(updatedSubject, updated(nil), anyFiles, watchesMain) if err != nil { t.Fatal(err) } @@ -86,7 +86,7 @@ func TestUpdatedBecomesAPullMessage(t *testing.T) { // 1235/AC2 (the conversion; } // 1235/AC3, merged into this test for the test budget: dirs not said are not known. - msg, err = Convert(updatedSubject, updated(map[string]any{"module_dirs": nil, "module_dirs_said": nil}), nil, anyFiles) + msg, err = Convert(updatedSubject, updated(map[string]any{"module_dirs": nil, "module_dirs_said": nil}), anyFiles, watchesMain) if err != nil { t.Fatal(err) } @@ -95,7 +95,7 @@ func TestUpdatedBecomesAPullMessage(t *testing.T) { // 1235/AC2 (the conversion; t.Errorf("dirs = %v known = %v, want not known", f.Dirs, f.DirsKnown) } // Said and empty is known and empty, not unknown. - msg, err = Convert(updatedSubject, updated(map[string]any{"module_dirs": []string{}, "module_dirs_said": true}), nil, anyFiles) + msg, err = Convert(updatedSubject, updated(map[string]any{"module_dirs": []string{}, "module_dirs_said": true}), anyFiles, watchesMain) if err != nil { t.Fatal(err) } @@ -104,7 +104,7 @@ func TestUpdatedBecomesAPullMessage(t *testing.T) { // 1235/AC2 (the conversion; t.Errorf("dirs = %v known = %v, want known and empty", f.Dirs, f.DirsKnown) } // Dirs listed without the forge saying it could read them are not trusted. - msg, err = Convert(updatedSubject, updated(map[string]any{"module_dirs_said": nil}), nil, anyFiles) + msg, err = Convert(updatedSubject, updated(map[string]any{"module_dirs_said": nil}), anyFiles, watchesMain) if err != nil { t.Fatal(err) } @@ -114,7 +114,7 @@ func TestUpdatedBecomesAPullMessage(t *testing.T) { // 1235/AC2 (the conversion; } func TestClosedBecomesAClosedPullMessage(t *testing.T) { // 1235/AC4 - msg, err := Convert(closedSubject, closed(), nil, anyFiles) + msg, err := Convert(closedSubject, closed(), anyFiles, watchesMain) if err != nil { t.Fatal(err) } @@ -130,17 +130,17 @@ func TestClosedBecomesAClosedPullMessage(t *testing.T) { // 1235/AC4 if !forgemsg.SameForge(f, want) { t.Errorf("message = %+v, want %+v", f, want) } - // Issue 1422, ADR 0484 §11: a head pushed to a pull request into another base is written closed, read nothing for, and said; - // one into main is no closed message, and an event naming no base is written as before. - msg, err = Convert(updatedSubject, updated(map[string]any{"base": "nox-mesh"}), nil, nil) + // Issue 1422; design 47 G §507: a head pushed to a pull request into a base no registration watches is written closed, + // read nothing for, and said; one into a watched branch is no closed message, and an event naming no base is written as before. + msg, err = Convert(updatedSubject, updated(map[string]any{"base": "nox-mesh"}), nil, watchesMain) if err != nil { t.Fatal(err) } - if f, err = forgemsg.DecodeForge(msg.Data); err != nil || !forgemsg.SameForge(f, want) || !strings.Contains(msg.Said, "is into nox-mesh, not main") { + if f, err = forgemsg.DecodeForge(msg.Data); err != nil || !forgemsg.SameForge(f, want) || !strings.Contains(msg.Said, "is into nox-mesh, which is not watched") { t.Errorf("an update into nox-mesh is %+v (%v), said %q; want it closed and said", f, err, msg.Said) } for _, base := range []any{"main", nil} { - msg, err = Convert(updatedSubject, updated(map[string]any{"base": base}), nil, anyFiles) + msg, err = Convert(updatedSubject, updated(map[string]any{"base": base}), anyFiles, watchesMain) if f, _ = forgemsg.DecodeForge(msg.Data); err != nil || f.Action != forgemsg.PullUpdated || msg.Said != "" { t.Errorf("an update with base %v is %+v (%v), said %q; want it updated", base, f, err, msg.Said) } @@ -165,7 +165,7 @@ func TestUnreadableEventsWriteNothing(t *testing.T) { // 1235/AC5 "closed no head": {closedSubject, mustDrop(closed(), "head"), "head"}, "closed no repo": {closedSubject, mustDrop(closed(), "repository"), "repo"}, } { - msg, err := Convert(c.subject, c.data, nil, anyFiles) + msg, err := Convert(c.subject, c.data, anyFiles, watchesMain) if err == nil { t.Errorf("%s: a message %q was made", name, msg.Event) continue @@ -192,7 +192,7 @@ func mustDrop(body []byte, key string) []byte { func TestOtherSubjectsAreNotHandled(t *testing.T) { // 1235, 1236 for _, s := range []string{"mesh.seat.git.event.something.else", "mesh.mod.gitea.event.pull.updated", "mesh.mod.other.event.forge.delete"} { - if _, err := Convert(s, updated(nil), nil, anyFiles); !errors.Is(err, ErrNotHandled) { + if _, err := Convert(s, updated(nil), anyFiles, watchesMain); !errors.Is(err, ErrNotHandled) { t.Errorf("%s: err = %v, want ErrNotHandled", s, err) } } @@ -202,23 +202,23 @@ func TestOtherSubjectsAreNotHandled(t *testing.T) { // 1235, 1236 b, _ := json.Marshal(map[string]any{"repository": "novox/mesh-sdk", "ref": ref, "ref-type": refType, "sender": "jochen"}) return b } - msg, err := Convert(DeletedSubject, del("branch", "feat/x"), nil, anyFiles) + msg, err := Convert(DeletedSubject, del("branch", "feat/x"), anyFiles, watchesMain) var bd BranchDeleted if err != nil || msg.Event != "forge.novox.mesh-sdk.branch-deleted" || json.Unmarshal(msg.Data, &bd) != nil || bd.Kind != ForgeBranchDeleted || bd.Repository != "novox/mesh-sdk" || bd.Branch != "feat/x" || bd.Head != "" || msg.Said == "" { t.Errorf("forge.delete of a branch = %+v (%s), %v", msg, msg.Data, err) } for _, c := range []struct{ refType, ref string }{{"tag", "go/v0.1.13"}, {"branch", "main"}} { - if _, err := Convert(DeletedSubject, del(c.refType, c.ref), nil, anyFiles); !errors.Is(err, ErrNotHandled) { + if _, err := Convert(DeletedSubject, del(c.refType, c.ref), anyFiles, watchesMain); !errors.Is(err, ErrNotHandled) { t.Errorf("forge.delete of %s %s: err = %v, want ErrNotHandled", c.refType, c.ref, err) } } - if _, err := Convert(DeletedSubject, []byte(`{"ref":"x","ref-type":"branch"}`), nil, anyFiles); !errors.Is(err, ErrUnreadable) { + if _, err := Convert(DeletedSubject, []byte(`{"ref":"x","ref-type":"branch"}`), anyFiles, watchesMain); !errors.Is(err, ErrUnreadable) { t.Errorf("forge.delete naming no repository: err = %v, want ErrUnreadable", err) } // 1236: a merged event without a full merge commit is unreadable, and the forge is not asked. var calls []call - _, err = Convert(mergedSubject, mergedBody(map[string]any{"merge_commit_sha": "abc"}), treesOf(&calls, "r0", map[string]string{"a": "t"}, nil), anyFiles) + _, err = convertMerged(mergedBody(map[string]any{"merge_commit_sha": "abc"}), treesOf(&calls, "r0", map[string]string{"a": "t"}, nil), anyFiles, nil) if !errors.Is(err, ErrUnreadable) || len(calls) != 0 { t.Errorf("merged without a merge commit: err = %v calls = %v", err, calls) } @@ -247,7 +247,7 @@ func treesOf(calls *[]call, root string, trees map[string]string, err error) Tre func TestMergeIntoMainBecomesAMergedMessage(t *testing.T) { // 1236/AC1 var calls []call answer := map[string]string{"modules/a": "t1", "modules/b": "t2"} - msg, err := Convert(mergedSubject, mergedBody(nil), treesOf(&calls, "r0", answer, nil), anyFiles) + msg, err := convertMerged(mergedBody(nil), treesOf(&calls, "r0", answer, nil), anyFiles, nil) if err != nil { t.Fatal(err) } @@ -259,7 +259,7 @@ func TestMergeIntoMainBecomesAMergedMessage(t *testing.T) { // 1236/AC1 t.Fatal(err) } want := forgemsg.Forge{V: keys.Version, Kind: forgemsg.ForgeMerged, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, - Head: head, Merge: merge, Root: "r0", Trees: answer, Manifests: map[string]string{"modules/a": "{}", "modules/b": "{}"}} + Head: head, Merge: merge, Base: "main", Root: "r0", Trees: answer, Manifests: map[string]string{"modules/a": "{}", "modules/b": "{}"}} if !forgemsg.SameForge(f, want) { t.Errorf("message = %+v, want %+v", f, want) } @@ -268,9 +268,14 @@ func TestMergeIntoMainBecomesAMergedMessage(t *testing.T) { // 1236/AC1 } } -func TestMergeIntoAnotherBranchIsAClosedPull(t *testing.T) { // 1236/AC2 +// 1236/AC2; design 47 G §509, §714: a pull request merged into a branch no registration watches is a closed pull, and one +// merged into a watched branch writes nothing here: its one message is its branch's move (MoveMessage). +func TestMergeIntoAnotherBranchIsAClosedPull(t *testing.T) { var calls []call - msg, err := Convert(mergedSubject, mergedBody(map[string]any{"base": "feat/y"}), treesOf(&calls, "r0", map[string]string{"a": "t"}, nil), anyFiles) + if _, err := Convert(mergedSubject, mergedBody(nil), anyFiles, watchesMain); !errors.Is(err, ErrNotHandled) { + t.Errorf("a merge into the watched main: %v, want nothing written (ErrNotHandled)", err) + } + msg, err := Convert(mergedSubject, mergedBody(map[string]any{"base": "feat/y"}), anyFiles, watchesMain) if err != nil { t.Fatal(err) } @@ -290,7 +295,7 @@ func TestMergeIntoAnotherBranchIsAClosedPull(t *testing.T) { // 1236/AC2 func TestTreesRefusedWritesNothingAndSaysWhy(t *testing.T) { // 1236/AC3 (the conversion; the delay is in main_test) var calls []call - msg, err := Convert(mergedSubject, mergedBody(nil), treesOf(&calls, "", nil, errors.New("the listing for novox/mesh-delivery was cut")), anyFiles) + msg, err := convertMerged(mergedBody(nil), treesOf(&calls, "", nil, errors.New("the listing for novox/mesh-delivery was cut")), anyFiles, nil) if !errors.Is(err, ErrRetry) || msg.Data != nil { t.Fatalf("err = %v data = %s, want ErrRetry and no message", err, msg.Data) } @@ -305,13 +310,13 @@ func TestTreesRefusedWritesNothingAndSaysWhy(t *testing.T) { // 1236/AC3 (the co // message with no trees, so the indexer takes it in and the repository's reach is not held for ever. func TestNoModuleJSONAnywhereWritesAMergedMessageWithNoTrees(t *testing.T) { var calls []call - msg, err := Convert(mergedSubject, mergedBody(nil), treesOf(&calls, "r0", map[string]string{}, nil), anyFiles) + msg, err := convertMerged(mergedBody(nil), treesOf(&calls, "r0", map[string]string{}, nil), anyFiles, nil) if err != nil { t.Fatal(err) } f, err := forgemsg.DecodeForge(msg.Data) want := forgemsg.Forge{V: keys.Version, Kind: forgemsg.ForgeMerged, Repository: "novox/mesh-delivery", Branch: "feat/x", Number: 12, - Head: head, Merge: merge, Root: "r0"} + Head: head, Merge: merge, Base: "main", Root: "r0"} if err != nil || msg.Event != "forge.novox.mesh-delivery.merged" || !forgemsg.SameForge(f, want) || !strings.Contains(msg.Said, "no module.json") { t.Errorf("event %q message %+v said %q, %v; want a merged message with no trees", msg.Event, f, msg.Said, err) } @@ -330,7 +335,7 @@ func openEntry(repo string, number int, base, branch, sha string) map[string]any func TestOpenPullsAreUpdatedPullMessagesWithUnknownDirs(t *testing.T) { // 1237/AC1 h2 := strings.Repeat("a", 40) - msgs, skipped, err := OpenPulls(openAnswer(openEntry("mesh-delivery", 12, "main", "feat/x", head), openEntry("mesh-catalog", 300, "main", "feat/x", h2))) + msgs, skipped, err := OpenPulls(openAnswer(openEntry("mesh-delivery", 12, "main", "feat/x", head), openEntry("mesh-catalog", 300, "main", "feat/x", h2)), watchesMain) if err != nil || len(skipped) != 0 || len(msgs) != 2 { t.Fatalf("msgs %d skipped %v err %v", len(msgs), skipped, err) } @@ -353,15 +358,15 @@ func TestOpenPullsAreUpdatedPullMessagesWithUnknownDirs(t *testing.T) { // 1237/ func TestOpenPullsUnreadableEntryIsNamedAndTheRestGo(t *testing.T) { // 1237, 1237/AC4 // 1237/AC4: an open pull request into another branch gets no message. - if msgs, skipped, err := OpenPulls(openAnswer(openEntry("mesh-delivery", 12, "feat/y", "feat/x", head))); err != nil || len(msgs) != 0 || len(skipped) != 0 { + if msgs, skipped, err := OpenPulls(openAnswer(openEntry("mesh-delivery", 12, "feat/y", "feat/x", head)), watchesMain); err != nil || len(msgs) != 0 || len(skipped) != 0 { t.Errorf("into another branch: msgs %d skipped %v err %v, want none", len(msgs), skipped, err) } - msgs, skipped, err := OpenPulls(openAnswer(openEntry("a", 1, "main", "b", "short"), openEntry("mesh-delivery", 12, "main", "feat/x", head))) + msgs, skipped, err := OpenPulls(openAnswer(openEntry("a", 1, "main", "b", "short"), openEntry("mesh-delivery", 12, "main", "feat/x", head)), watchesMain) if err != nil || len(msgs) != 1 || len(skipped) != 1 || !strings.Contains(skipped[0], "novox/a#1") { t.Errorf("msgs %d skipped %v err %v", len(msgs), skipped, err) } for _, bad := range []string{`{"pulls":null}`, `{}`, `[]`, `x`, string(gitFixture(t, "open-pulls answers an error, never none", "verb:open-pulls.answer"))} { - if _, _, err := OpenPulls([]byte(bad)); err == nil { + if _, _, err := OpenPulls([]byte(bad), watchesMain); err == nil { t.Errorf("%s was taken as an answer", bad) } } @@ -371,7 +376,7 @@ 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) + m, err := OpenMessage(answer, watchesMain) if err != nil || m.Event != "forge.open" { t.Fatalf("message %+v err %v", m, err) } @@ -382,12 +387,12 @@ func TestOpenMessageNamesEveryOpenPullIntoTheTrunk(t *testing.T) { // 1279/AC6 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":[]}`)) + m, err = OpenMessage([]byte(`{"pulls":[]}`), watchesMain) if f, derr := forgemsg.DecodeForge(m.Data); err != nil || derr != nil || f.Kind != forgemsg.ForgeOpen || len(f.Open) != 0 { t.Errorf("empty answer: %+v %v %v", f, err, derr) } for _, bad := range []string{`{"pulls":null}`, `{}`, `x`, `{"error":"the forge answered 502","pulls":[]}`} { - if _, err := OpenMessage([]byte(bad)); err == nil { + if _, err := OpenMessage([]byte(bad), watchesMain); err == nil { t.Errorf("%s was taken as an answer", bad) } } @@ -490,3 +495,25 @@ func slicesEqual(a, b []string) bool { } return true } + +// watchesMain watches the branch main of every repository, as the registrations of a test name it. +var watchesMain = watchedOf("main") + +// watchedOf watches the branches of every repository, before the first registration, as the gate's source names them. +func watchedOf(branches ...string) *Watched { + w := &Watched{} + w.Set(nil, branches) + return w +} + +// convertMerged is the merged message of a pull.merged body as the move of its base writes it (landed, MoveMessage), so the +// tests of what a merge reads keep their bodies. +func convertMerged(data []byte, trees Trees, files Files, kept Kept) (Message, error) { + p, err := pullOf(data) + if err != nil { + return Message{}, unreadable(MergedSubject, "%v", err) + } + f := forgemsg.Forge{Kind: forgemsg.ForgeMerged, Repository: p.Owner + "/" + p.Repo, Branch: p.Head, Number: p.Number, + Head: p.HeadSHA, Merge: p.MergeSHA, Base: p.Base} + return landed(MergedSubject, f, p.MergeSHA, trees, files, kept) +} diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/call.go b/modules/mesh-delivery-forge-bridge/internal/bridge/call.go index 012fdea5..4c51d447 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/call.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/call.go @@ -8,6 +8,7 @@ import ( "strings" git "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/git" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" ) // Caller calls a seat's verb by its manifest key (`seat:.`) with a JSON body and returns its answer: the @@ -87,15 +88,17 @@ func FilesVia(ctx context.Context, call Caller) Files { // errNoAnswer is a read whose answer carries nothing the contract says it must. var errNoAnswer = errors.New("the seat git's answer carries nothing") -// MergedMessage is the merged message of a merge into the trunk, asked again: built as the bridge builds the one it writes -// on FORGE (it is that event's body through ConvertKept), from the pull request's owner, repository, number, branch, head -// and merge commit, which the seat git's verb pull answers. It reads the forge for every module directory (no kept inputs). -func MergedMessage(owner, repo string, number int, branch, head, merge string, trees Trees, files Files) (Message, error) { - body, err := json.Marshal(git.PullMergedEvent{Repository: repository(owner, repo), Number: int64(number), Branch: branch, Head: head, Base: Trunk, MergeCommit: merge}) - if err != nil { - return Message{}, err +// MergedMessage is the merged message of a merge into the branch base, asked again: built as the bridge builds the one +// it writes on FORGE for the branch's move (MoveMessage), from the pull request's owner, repository, number, branch, head, +// merge commit and base, which the seat git's verb pull answers. It reads the forge for every module directory (no kept +// inputs). +func MergedMessage(owner, repo string, number int, branch, head, merge, base string, trees Trees, files Files) (Message, error) { + if number <= 0 || branch == "" || base == "" || !sha40.MatchString(head) { + return Message{}, unreadable("bridge_merged_message", "%s#%d names no branch, base or 40-hex head", repository(owner, repo), number) } - return ConvertKept(MergedSubject, body, trees, files, nil) + f := forgemsg.Forge{Kind: forgemsg.ForgeMerged, Repository: repository(owner, repo), Branch: branch, Number: number, Head: head, + Merge: merge, Base: base} + return landed(repository(owner, repo)+"#"+fmt.Sprint(number), f, merge, trees, files, nil) } // splitRepository is a `/` cut in two, refusing anything else. diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go b/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go index aab23cd9..09b4418e 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go @@ -66,7 +66,7 @@ func TestMergedCarriesManifestsAndLocks(t *testing.T) { // 1448/AC1 manifest, sum := real(t, "module.json"), real(t, "go.sum") forge := &forgeFiles{have: map[string][]byte{"modules/a/module.json": manifest, "modules/a/go.sum": sum, "modules/b/module.json": []byte(`{"module":"b"}`)}} var calls []call - msg, err := Convert(mergedSubject, mergedBody(nil), treesOf(&calls, "r0", map[string]string{"modules/a": "t1", "modules/b": "t2"}, nil), forge.get) + msg, err := convertMerged(mergedBody(nil), treesOf(&calls, "r0", map[string]string{"modules/a": "t1", "modules/b": "t2"}, nil), forge.get, nil) if err != nil { t.Fatal(err) } @@ -87,7 +87,7 @@ func TestMergedCarriesManifestsAndLocks(t *testing.T) { // 1448/AC1 // ADR 0567: assignments is the seat mesh-module-assigner's: a merge of the catalogue carries no assignments file and asks the // forge for none. forge.calls = nil - msg, err = Convert(mergedSubject, mergedBody(map[string]any{"repo": "mesh-catalog"}), treesOf(&calls, "r0", map[string]string{"modules/a": "t1"}, nil), forge.get) + msg, err = convertMerged(mergedBody(map[string]any{"repo": "mesh-catalog"}), treesOf(&calls, "r0", map[string]string{"modules/a": "t1"}, nil), forge.get, nil) if err != nil { t.Fatal(err) } @@ -108,19 +108,19 @@ func TestMergedCarriesManifestsAndLocks(t *testing.T) { // 1448/AC1 t.Errorf("a merge of the catalogue with no predefined.json does not say so: %s", msg.Data) } forge.have[PredefinedFile] = []byte(`{"font-size":{"type":"integer","value":13}}`) - msg, _ = Convert(mergedSubject, mergedBody(map[string]any{"repo": "mesh-catalog"}), treesOf(&calls, "r0", map[string]string{"modules/a": "t1"}, nil), forge.get) + msg, _ = convertMerged(mergedBody(map[string]any{"repo": "mesh-catalog"}), treesOf(&calls, "r0", map[string]string{"modules/a": "t1"}, nil), forge.get, nil) if f, err = forgemsg.DecodeForge(msg.Data); err != nil || string(f.Predefined) != string(forge.have[PredefinedFile]) || f.PredefinedRead { t.Errorf("predefined = %s (%v)", f.Predefined, err) } forge.calls = nil - msg, _ = Convert(mergedSubject, mergedBody(nil), treesOf(&calls, "r0", map[string]string{"modules/a": "t1"}, nil), forge.get) + msg, _ = convertMerged(mergedBody(nil), treesOf(&calls, "r0", map[string]string{"modules/a": "t1"}, nil), forge.get, nil) for _, c := range forge.calls { if c.path == PredefinedFile { t.Errorf("another repository's merge asked for %s", c.path) } } forge.errs = map[string]error{PredefinedFile: errors.New("forge down")} - if _, err := Convert(mergedSubject, mergedBody(map[string]any{"repo": "mesh-catalog"}), treesOf(&calls, "r0", map[string]string{"modules/a": "t1"}, nil), forge.get); !errors.Is(err, ErrRetry) { + if _, err := convertMerged(mergedBody(map[string]any{"repo": "mesh-catalog"}), treesOf(&calls, "r0", map[string]string{"modules/a": "t1"}, nil), forge.get, nil); !errors.Is(err, ErrRetry) { t.Errorf("a forge that cannot say is %v, want a retry", err) } // ADR 0632 §2: of a manifest naming a template, the template is read at the merge and the module's needs carried @@ -130,7 +130,7 @@ func TestMergedCarriesManifestsAndLocks(t *testing.T) { // 1448/AC1 forge.have["modules/d/module.json"] = []byte(`{"module":"d","version":"1","resources":[{"id":"daemon","type":"file",` + `"path":"/etc/d.json","template":"files/daemon.json"},{"id":"gone","type":"file","path":"/etc/g","template":"files/gone"}]}`) forge.have["modules/d/files/daemon.json"] = []byte(`{"registry":"${seat:container-image-registry:reach}"}`) - msg, err = Convert(mergedSubject, mergedBody(nil), treesOf(&calls, "r0", map[string]string{"modules/b": "t2", "modules/d": "t4"}, nil), forge.get) + msg, err = convertMerged(mergedBody(nil), treesOf(&calls, "r0", map[string]string{"modules/b": "t2", "modules/d": "t4"}, nil), forge.get, nil) if err != nil { t.Fatal(err) } @@ -144,7 +144,7 @@ func TestMergedCarriesManifestsAndLocks(t *testing.T) { // 1448/AC1 func TestPullWithKnownDirsCarriesTheSameAtTheHead(t *testing.T) { // 1448/AC2 manifest := real(t, "module.json") forge := &forgeFiles{have: map[string][]byte{"modules/a/module.json": manifest, "modules/a/go.sum": []byte("git.novox.be/novox/mesh-sdk/go v0.1.7 h1:x=\n")}} - msg, err := Convert(updatedSubject, updated(nil), nil, forge.get) + msg, err := Convert(updatedSubject, updated(nil), forge.get, watchesMain) if err != nil { t.Fatal(err) } @@ -157,7 +157,7 @@ func TestPullWithKnownDirsCarriesTheSameAtTheHead(t *testing.T) { // 1448/AC2 } // Dirs not known: neither field, and the forge is not asked. forge = &forgeFiles{} - msg, err = Convert(updatedSubject, updated(map[string]any{"module_dirs": nil, "module_dirs_said": nil}), nil, forge.get) + msg, err = Convert(updatedSubject, updated(map[string]any{"module_dirs": nil, "module_dirs_said": nil}), forge.get, watchesMain) if err != nil { t.Fatal(err) } @@ -176,17 +176,17 @@ func TestAForgeErrorOrAMissingManifestIsRetryAndALockNotFoundIsLeftOut(t *testin "a lock file error": {have: map[string][]byte{"modules/a/module.json": []byte("{}")}, errs: map[string]error{"modules/a/go.sum": boom}}, "a second lock error": {have: map[string][]byte{"modules/a/module.json": []byte("{}")}, errs: map[string]error{"modules/a/package-lock.json": boom}}, } { - msg, err := Convert(mergedSubject, mergedBody(nil), trees(), forge.get) + msg, err := convertMerged(mergedBody(nil), trees(), forge.get, nil) if !errors.Is(err, ErrRetry) || msg.Data != nil { t.Errorf("%s: err = %v data = %s, want ErrRetry and no message", name, err, msg.Data) } - _, err = Convert(updatedSubject, updated(nil), nil, forge.get) + _, err = Convert(updatedSubject, updated(nil), forge.get, watchesMain) if !errors.Is(err, ErrRetry) { t.Errorf("%s (pull): err = %v, want ErrRetry", name, err) } } forge := &forgeFiles{have: map[string][]byte{"modules/a/module.json": []byte("{}")}} - msg, err := Convert(mergedSubject, mergedBody(nil), trees(), forge.get) + msg, err := convertMerged(mergedBody(nil), trees(), forge.get, nil) if err != nil { t.Fatal(err) } @@ -198,7 +198,7 @@ func TestAForgeErrorOrAMissingManifestIsRetryAndALockNotFoundIsLeftOut(t *testin func TestTheFilesAreAskedAtTheMergeCommitByOwnerRepoPathAndRef(t *testing.T) { // 1448/AC4 (the call's arguments; the manifest is in manifest_test) forge := &forgeFiles{have: map[string][]byte{"modules/a/module.json": []byte("{}")}} var calls []call - if _, err := Convert(mergedSubject, mergedBody(nil), treesOf(&calls, "r0", map[string]string{"modules/a": "t1"}, nil), forge.get); err != nil { + if _, err := convertMerged(mergedBody(nil), treesOf(&calls, "r0", map[string]string{"modules/a": "t1"}, nil), forge.get, nil); err != nil { t.Fatal(err) } want := []fileCall{{"novox", "mesh-delivery", "modules/a/module.json", merge}, {"novox", "mesh-delivery", "modules/a/go.sum", merge}, @@ -335,7 +335,7 @@ func TestOneFailingFileIsRetryAndEveryCallHasReturnedBeforeTheReadDoes(t *testin dirsMap[d] = "t" } slow2 := &slowForge{failOn: "modules/m05/module.json"} - if _, err := Convert(mergedSubject, mergedBody(nil), treesOf(&calls, "r0", dirsMap, nil), slow2.get); !errors.Is(err, ErrRetry) { + if _, err := convertMerged(mergedBody(nil), treesOf(&calls, "r0", dirsMap, nil), slow2.get, nil); !errors.Is(err, ErrRetry) { t.Errorf("err = %v, want ErrRetry", err) } } @@ -353,7 +353,7 @@ func mergeReadsOnlyChangedDirs(t *testing.T) { forge := &forgeFiles{have: map[string][]byte{"modules/a/module.json": manifest, "modules/a/go.sum": sum, "modules/b/module.json": []byte(`{"module":"b"}`)}} kept := memKept{} var calls []call - first, err := ConvertKept(mergedSubject, mergedBody(nil), treesOf(&calls, "r0", map[string]string{"modules/a": "t1", "modules/b": "t2"}, nil), forge.get, kept) + first, err := convertMerged(mergedBody(nil), treesOf(&calls, "r0", map[string]string{"modules/a": "t1", "modules/b": "t2"}, nil), forge.get, kept) if err != nil { t.Fatal(err) } @@ -364,7 +364,7 @@ func mergeReadsOnlyChangedDirs(t *testing.T) { forge.calls = nil forge.have["modules/b/module.json"] = []byte(`{"module":"b","v":2}`) trees := map[string]string{"modules/a": "t1", "modules/b": "t3"} - next, err := ConvertKept(mergedSubject, mergedBody(nil), treesOf(&calls, "r1", trees, nil), forge.get, kept) + next, err := convertMerged(mergedBody(nil), treesOf(&calls, "r1", trees, nil), forge.get, kept) if err != nil { t.Fatal(err) } @@ -376,7 +376,7 @@ func mergeReadsOnlyChangedDirs(t *testing.T) { if len(forge.calls) != 3 { t.Errorf("the second merge read %d files, want 3", len(forge.calls)) } - whole, err := Convert(mergedSubject, mergedBody(nil), treesOf(&calls, "r1", trees, nil), forge.get) + whole, err := convertMerged(mergedBody(nil), treesOf(&calls, "r1", trees, nil), forge.get, nil) if err != nil { t.Fatal(err) } @@ -390,7 +390,7 @@ func mergeReadsOnlyChangedDirs(t *testing.T) { // rest: a merge too large for one try's bound converges over its tries. kept2 := memKept{} failing := &forgeFiles{have: forge.have, errs: map[string]error{"modules/b/module.json": errors.New("context deadline exceeded")}} - if _, err := ConvertKept(mergedSubject, mergedBody(nil), treesOf(&calls, "r1", trees, nil), failing.get, kept2); !errors.Is(err, ErrRetry) { + if _, err := convertMerged(mergedBody(nil), treesOf(&calls, "r1", trees, nil), failing.get, kept2); !errors.Is(err, ErrRetry) { t.Fatalf("err = %v, want ErrRetry", err) } if _, ok := kept2["t1"]; !ok || len(kept2) != 1 { // modules/a's files are asked first, before the failure is seen @@ -398,7 +398,7 @@ func mergeReadsOnlyChangedDirs(t *testing.T) { } failing.calls = nil failing.errs = nil - again, err := ConvertKept(mergedSubject, mergedBody(nil), treesOf(&calls, "r1", trees, nil), failing.get, kept2) + again, err := convertMerged(mergedBody(nil), treesOf(&calls, "r1", trees, nil), failing.get, kept2) if err != nil { t.Fatal(err) } diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/forgemade.go b/modules/mesh-delivery-forge-bridge/internal/bridge/forgemade.go index 487fa233..6cc982b7 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/forgemade.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/forgemade.go @@ -13,12 +13,13 @@ type ForgePull struct { Number int Branch string // the pull request's head branch Head string // the head commit - Merge string // the merge commit into the trunk + Merge string // the merge commit + Base string // the branch it was merged into } // ForgeMade reads, through the seat `git`'s verb pull, the merged pull request whose merge commit is merge: by number -// when number is given (its merge commit must then equal merge), else by the merge commit. A pull request not merged, -// merged into another branch than the trunk, or merged at another commit is refused. +// when number is given (its merge commit must then equal merge), else by the merge commit. A pull request not merged, or +// merged at another commit, is refused. func ForgeMade(ctx context.Context, call Caller, owner, repo string, number int, merge string) (ForgePull, error) { in := git.PullInput{Repository: repository(owner, repo), MergeCommit: merge} if number > 0 { @@ -36,10 +37,10 @@ func ForgeMade(ctx context.Context, call Caller, owner, repo string, number int, return ForgePull{}, fmt.Errorf("the forge says the pull request %s#%d is %s, not merged", in.Repository, p.Number, p.State) case p.MergeCommit != merge: return ForgePull{}, fmt.Errorf("the forge says %s#%d was merged at %s, not at %s", in.Repository, p.Number, p.MergeCommit, merge) - case p.Base != Trunk: - return ForgePull{}, fmt.Errorf("the forge says %s#%d was merged into %s, not the trunk %s", in.Repository, p.Number, p.Base, Trunk) + case p.Base == "": + return ForgePull{}, fmt.Errorf("the forge gives no base branch for %s#%d", in.Repository, p.Number) case p.Branch == "": return ForgePull{}, fmt.Errorf("the forge gives no head branch for %s#%d", in.Repository, p.Number) } - return ForgePull{Number: int(p.Number), Branch: p.Branch, Head: p.Head, Merge: p.MergeCommit}, nil + return ForgePull{Number: int(p.Number), Branch: p.Branch, Head: p.Head, Merge: p.MergeCommit, Base: p.Base}, nil } diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/moved.go b/modules/mesh-delivery-forge-bridge/internal/bridge/moved.go new file mode 100644 index 00000000..a559662c --- /dev/null +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/moved.go @@ -0,0 +1,95 @@ +package bridge + +import ( + "context" + "encoding/json" + "fmt" + "strings" + + git "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/git" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" + "git.novox.be/novox/mesh-sdk/go/keys" +) + +// BranchMovedSubject is the seat git's branch-moved (design 47 G §11): a branch's new head, which a merge and an admitted +// push alike say, kept on the seat's stream SEAT_git_branch-moved and heard there through a durable consumer. +const BranchMovedSubject = git.SubjectBranchMoved + +// Moved is a branch-moved as the bridge reads it: the repository's two halves and the branch. The head the event says is +// never taken: the commit is read from the forge's own record by the seat's verb branch (§11). +type Moved struct { + Owner, Repo, Branch string +} + +// MovedOf reads a branch-moved's body. +func MovedOf(data []byte) (Moved, error) { + var e git.BranchMovedEvent + if err := json.Unmarshal(data, &e); err != nil { + return Moved{}, unreadable(BranchMovedSubject, "the body is not json: %v", err) + } + m := Moved{Branch: e.Branch} + m.Owner, m.Repo, _ = strings.Cut(e.Repository, "/") + if m.Owner == "" || m.Repo == "" || strings.ContainsAny(m.Owner+m.Repo, "./ *>") || m.Branch == "" { + return Moved{}, unreadable(BranchMovedSubject, "no repository / or branch (%q, %q)", e.Repository, e.Branch) + } + return m, nil +} + +// Repository is `/`. +func (m Moved) Repository() string { return repository(m.Owner, m.Repo) } + +// HeadKey is the key of a watched branch's last head written in the state heads: `branch...` (A-236), apart from a pull request's `..`. +func (m Moved) HeadKey() string { + return "branch." + m.Owner + "." + m.Repo + "." + keys.EscapeName(m.Branch) +} + +// Branch is a branch as the seat git's verb branch reads it: its head and whether it is protected. +type Branch struct { + Head string + Protected bool +} + +// ReadBranch reads a branch's head and protection through the seat git's verb branch (§11, §17); an answer that says +// an error, or names no 40-hex head, is an error, never a branch with none. +func ReadBranch(ctx context.Context, call Caller, m Moved) (Branch, error) { + a, err := askGit[git.BranchAnswer](ctx, call, git.VerbBranch, git.BranchInput{Repository: m.Repository(), Branch: m.Branch}) + if err != nil { + return Branch{}, err + } + if !sha40.MatchString(a.Head) { + return Branch{}, fmt.Errorf("the seat git's branch %s of %s names no 40-hex head (%q)", m.Branch, m.Repository(), a.Head) + } + return Branch{Head: a.Head, Protected: a.Protected}, nil +} + +// MoveMessage is the one message of a move of a watched branch to head (design 47 G §509, §714, A-236): merged, with +// the pull request and the watched branch as base, when the seat's verb pull names head a pull request's merge; pushed, +// with the branch, head and before (the head last written for the branch, empty for none), when it names none. Both +// carry the take-in inputs at head. A pull the forge could not say is ErrRetry, never read as none. +func MoveMessage(ctx context.Context, call Caller, m Moved, head, before string, trees Trees, files Files, kept Kept) (Message, error) { + what := fmt.Sprintf("%s of %s at %s", m.Branch, m.Repository(), head) + raw, err := call(ctx, GitKey(git.VerbPull), git.PullInput{Repository: m.Repository(), MergeCommit: head}) + if err != nil { + return Message{}, fmt.Errorf("%w: %s: the seat git did not answer pull: %v", ErrRetry, what, err) + } + var a git.PullAnswer + if err := json.Unmarshal(raw, &a); err != nil { + return Message{}, fmt.Errorf("%w: %s: the seat git's answer of pull is not its contract's: %v", ErrRetry, what, err) + } + f := forgemsg.Forge{Repository: m.Repository()} + switch { + case a.Error != "" && a.ErrorKind == git.ErrorKindNotFound: + f.Kind, f.Branch, f.Head, f.Before = forgemsg.ForgePushed, m.Branch, head, before + case a.Error != "": + return Message{}, fmt.Errorf("%w: %s: the seat git refused pull: %s", ErrRetry, what, a.Error) + case a.Pull == nil || a.Pull.Number <= 0 || a.Pull.MergeCommit != head || a.Pull.State != git.PullStateMerged: + return Message{}, fmt.Errorf("%w: %s: the seat git's pull at that merge commit names no merged pull request merged there", ErrRetry, what) + default: + if !sha40.MatchString(a.Pull.Head) || a.Pull.Branch == "" { + return Message{}, unreadable(BranchMovedSubject, "%s: the pull request %s#%d names no branch or 40-hex head", what, m.Repository(), a.Pull.Number) + } + f.Kind, f.Branch, f.Number, f.Head, f.Merge, f.Base = forgemsg.ForgeMerged, a.Pull.Branch, int(a.Pull.Number), a.Pull.Head, head, m.Branch + } + return landed(what, f, head, trees, files, kept) +} diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/moved_test.go b/modules/mesh-delivery-forge-bridge/internal/bridge/moved_test.go new file mode 100644 index 00000000..17ed5a68 --- /dev/null +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/moved_test.go @@ -0,0 +1,77 @@ +package bridge + +import ( + "context" + "encoding/json" + "errors" + "testing" + + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" + + "mesh-delivery-forge-bridge/internal/gitfixture" +) + +// Design 47 G §4, §11, §507, §509, §714, A-236 (FB1, FB9, FB10): the watched branches are the registrations' by origin, and +// before the first registration the gate's source's, no branch fixed in code; a move of a watched branch is one message: +// merged with its pull request and base when the seat's pull names the new head a pull request's merge, pushed with +// its branch, head and before when it names none, both carrying the commit's take-in inputs; a pull the forge could not +// say is heard again, never read as a push. +func TestAMoveOfAWatchedBranchIsOneMessage(t *testing.T) { + w := &Watched{} + w.Set([]Registered{{Origin: "novox/mesh-sdk", Branches: []string{"main", "beta"}}}, []string{"main"}) + for _, c := range []struct { + repo, branch string + want bool + }{{"novox/mesh-sdk", "beta", true}, {"novox/mesh-sdk", "feat/x", false}, {"novox/other", "main", false}} { + if got := w.Watches(c.repo, c.branch); got != c.want { + t.Errorf("registered: %s of %s watched %v, want %v", c.branch, c.repo, got, c.want) + } + } + if !watchedOf("main").Watches("novox/other", "main") || watchedOf("main").Watches("novox/other", "beta") { + t.Errorf("before the first registration the gate's source's branches are watched, of every repository") + } + + const landedAt, before = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", "cccccccccccccccccccccccccccccccccccccccc" + moved := Moved{Owner: "novox", Repo: "mesh-sdk", Branch: "beta"} + for _, c := range []struct { + name string + pull json.RawMessage + want forgemsg.Forge + retry bool + calls1 int + }{ + {"a merge", gitfixture.Given(t, "pull answers", "verb:pull.answer"), forgemsg.Forge{Kind: forgemsg.ForgeMerged, + Branch: "feat/47-G", Number: 178, Head: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", Merge: landedAt, Base: "beta"}, false, 1}, + {"an admitted push", json.RawMessage(`{"error":"no pull request merged at that commit","error-kind":"not-found"}`), + forgemsg.Forge{Kind: forgemsg.ForgePushed, Branch: "beta", Head: landedAt, Before: before}, false, 1}, + {"a pull the forge could not say", gitfixture.Given(t, "pull answers an error, never none", "verb:pull.answer"), + forgemsg.Forge{}, true, 0}, + } { + var asked []string + ask := func(_ context.Context, key string, body any) (json.RawMessage, error) { + b, _ := json.Marshal(body) + asked = append(asked, key+" "+string(b)) + return c.pull, nil + } + var trees []call + msg, err := MoveMessage(context.Background(), ask, moved, landedAt, before, treesOf(&trees, "r0", map[string]string{"modules/a": "t1"}, nil), anyFiles, nil) + if c.retry { + if !errors.Is(err, ErrRetry) || msg.Data != nil { + t.Errorf("%s: %v, %s; want ErrRetry and nothing written", c.name, err, msg.Data) + } + continue + } + if err != nil { + t.Fatalf("%s: %v", c.name, err) + } + f, err := forgemsg.DecodeForge(msg.Data) + c.want.V, c.want.Repository, c.want.Root = forgemsg.Version, "novox/mesh-sdk", "r0" + c.want.Trees, c.want.Manifests = map[string]string{"modules/a": "t1"}, map[string]string{"modules/a": "{}"} + if err != nil || !forgemsg.SameForge(f, c.want) || msg.Event != "forge.novox.mesh-sdk."+c.want.Kind { + t.Errorf("%s: %s %+v (%v), want %+v", c.name, msg.Event, f, err, c.want) + } + if len(trees) != c.calls1 || trees[0].sha != landedAt || asked[0] != `seat:git.pull {"merge-commit":"`+landedAt+`","repository":"novox/mesh-sdk"}` { + t.Errorf("%s: trees asked %v, pull asked %v; want the landed commit's", c.name, trees, asked) + } + } +} diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/paths_test.go b/modules/mesh-delivery-forge-bridge/internal/bridge/paths_test.go index c4ddac59..734224fd 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/paths_test.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/paths_test.go @@ -22,7 +22,7 @@ func TestAMergedMessageCarriesNoNamedPathsTheBridgeDerived(t *testing.T) { return nil, ErrNotFound } var calls []call - msg, err := Convert(mergedSubject, mergedBody(map[string]any{"repo": "mesh-sdk", "module_dirs": []string{""}}), treesOf(&calls, "r0", map[string]string{"": "r0"}, nil), files) + msg, err := convertMerged(mergedBody(map[string]any{"repo": "mesh-sdk", "module_dirs": []string{""}}), treesOf(&calls, "r0", map[string]string{"": "r0"}, nil), files, nil) if err != nil { t.Fatal(err) } diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/watched.go b/modules/mesh-delivery-forge-bridge/internal/bridge/watched.go new file mode 100644 index 00000000..27016255 --- /dev/null +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/watched.go @@ -0,0 +1,71 @@ +package bridge + +import ( + "slices" + "sync" +) + +// Watched is the branches the bridge watches, by repository (design 47 G §4, §11, §507, §532): the registrations' watched +// branches by their origin, from a first view of `registrations` and then `registration-changed`; before the first +// registration, the branches the gate's `source` answers, of every repository. No branch is fixed in code (§7). +type Watched struct { + mu sync.Mutex + byRepo map[string][]string + source []string + registered bool +} + +// Registered is one registration as the bridge reads it: its origin `/` and its watched branches. +type Registered struct { + Origin string + Branches []string +} + +// Set takes what the bridge read: the registrations (none before the first) and the gate's source's branches by name. +func (w *Watched) Set(regs []Registered, source []string) { + by := map[string][]string{} + for _, r := range regs { + for _, b := range r.Branches { + if b != "" && !slices.Contains(by[r.Origin], b) { + by[r.Origin] = append(by[r.Origin], b) + } + } + } + w.mu.Lock() + defer w.mu.Unlock() + w.byRepo, w.registered = by, len(regs) > 0 + w.source = slices.Clone(source) +} + +// Watches says whether a move of branch in repository is one the bridge writes. +func (w *Watched) Watches(repository, branch string) bool { + w.mu.Lock() + defer w.mu.Unlock() + if !w.registered { + return slices.Contains(w.source, branch) + } + return slices.Contains(w.byRepo[repository], branch) +} + +// Branches are the watched branches of repository, in the order the registrations name them; before the first +// registration, the gate's source's. +func (w *Watched) Branches(repository string) []string { + w.mu.Lock() + defer w.mu.Unlock() + if !w.registered { + return slices.Clone(w.source) + } + return slices.Clone(w.byRepo[repository]) +} + +// Repositories are the repositories the registrations name, sorted; none before the first registration. +func (w *Watched) Repositories() []string { + w.mu.Lock() + defer w.mu.Unlock() + out := make([]string, 0, len(w.byRepo)) + for r := range w.byRepo { + out = append(out, r) + } + slices.Sort(out) + return out +} diff --git a/modules/mesh-delivery-forge-bridge/manifest_test.go b/modules/mesh-delivery-forge-bridge/manifest_test.go index cdd979e1..be2ceab2 100644 --- a/modules/mesh-delivery-forge-bridge/manifest_test.go +++ b/modules/mesh-delivery-forge-bridge/manifest_test.go @@ -103,7 +103,9 @@ func TestManifestDeclarations(t *testing.T) { // 1235/AC1 // gate owes the forge*), never read from their state. The forge's events are the seat git's (design 47 *The seat git*); forge.delete: a branch's end, said as a // branch-deleted kind (design 51, DLV-FB1). The state performed's binding is asked again on the seat mesh-broker's // recreated and dropped, and asked of its show (design 47 G §577). - if want := []string{"git.forge-delete", "git.pull-closed", "git.pull-merged", "git.pull-updated", "mesh-broker.dropped", "mesh-broker.recreated", "mesh-delivery-checker.owed", "mesh-delivery.owed"}; !reflect.DeepEqual(sorted(m.Consumes), want) { + // Every move of a watched branch is heard on branch-moved, and the watched branches from the registrations, before the + // first the gate's source (design 47 G §4, §11, §507). + if want := []string{"git.branch-moved", "git.forge-delete", "git.pull-closed", "git.pull-merged", "git.pull-updated", "mesh-broker.dropped", "mesh-broker.recreated", "mesh-catalog.registration-changed", "mesh-delivery-checker.owed", "mesh-delivery.owed", "mesh-delivery.source-moved"}; !reflect.DeepEqual(sorted(m.Consumes), want) { t.Errorf("consumes = %v, want %v", m.Consumes, want) } // A group is read from the seat mesh-delivery's verb groups (design 47, acceptance 49): the bridge reads no state of @@ -111,8 +113,8 @@ func TestManifestDeclarations(t *testing.T) { // 1235/AC1 if len(m.Reads) != 0 { t.Errorf("reads = %v, want none", m.Reads) } - if want := []string{"seat:git.comment", "seat:git.file", "seat:git.merge-pull", "seat:git.note", "seat:git.open-pulls", - "seat:git.pull", "seat:git.status", "seat:git.tags", "seat:git.trees", "seat:git.view", "seat:mesh-broker.show", "seat:mesh-delivery.groups", "seat:mesh-hand-acts.hand-act"}; !reflect.DeepEqual(sorted(m.Invokes), want) { + if want := []string{"seat:git.branch", "seat:git.comment", "seat:git.file", "seat:git.merge-pull", "seat:git.note", "seat:git.open-pulls", + "seat:git.pull", "seat:git.status", "seat:git.tags", "seat:git.trees", "seat:git.view", "seat:mesh-broker.show", "seat:mesh-catalog.registrations", "seat:mesh-delivery.groups", "seat:mesh-delivery.source", "seat:mesh-hand-acts.hand-act"}; !reflect.DeepEqual(sorted(m.Invokes), want) { t.Errorf("invokes = %v, want %v", m.Invokes, want) } if b, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 || b != "${dir:state}/broker" { diff --git a/modules/mesh-delivery-forge-bridge/module.json b/modules/mesh-delivery-forge-bridge/module.json index a98c4328..0cbf07f2 100644 --- a/modules/mesh-delivery-forge-bridge/module.json +++ b/modules/mesh-delivery-forge-bridge/module.json @@ -7,6 +7,9 @@ "git.pull-merged", "git.pull-closed", "git.forge-delete", + "git.branch-moved", + "mesh-catalog.registration-changed", + "mesh-delivery.source-moved", "mesh-delivery.owed", "mesh-delivery-checker.owed", "mesh-broker.recreated", @@ -43,6 +46,9 @@ "seat:git.file", "seat:git.pull", "seat:git.tags", + "seat:git.branch", + "seat:mesh-catalog.registrations", + "seat:mesh-delivery.source", "seat:mesh-broker.show" ], "requires": [