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

FB1, FB9, FB10: the forge bridge writes one message per move of every watched branch

Browse Source 
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.<owner>.<name>.<escaped 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
This commit is contained in:
jschoubben
2026-11-12 09:15:04 +01:00
parent dcb7ba016c
commit dce4d01a58
22 changed files with 819 additions and 241 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main.go
+1 -1
View File
@@ -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)
}
modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main_test.go
+7 -3
View File
@@ -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,
modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go
+47 -19
View File
@@ -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<E>, SEAT_<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 {
modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go
+99 -40
View File
@@ -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
}
modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go
+72 -28
View File
@@ -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)
}
}
modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/moves.go
+143
View File
@@ -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
}
modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/open.go
+3 -3
View File
@@ -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)
}
modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/open_test.go
+1 -1
View File
@@ -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
modules/mesh-delivery-forge-bridge/forgemsg/forge.go
+40 -25
View File
@@ -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.<owner>.<name>.<kind>`.
@@ -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
modules/mesh-delivery-forge-bridge/forgemsg/forge_test.go
+8 -1
View File
@@ -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)
modules/mesh-delivery-forge-bridge/go.mod
+2
View File
@@ -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
modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go
+48 -54
View File
@@ -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})
modules/mesh-delivery-forge-bridge/internal/bridge/bridge_test.go
+57 -30
View File
@@ -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)
}
modules/mesh-delivery-forge-bridge/internal/bridge/call.go
+11 -8
View File
@@ -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:<seat>.<verb>`) 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 `<owner>/<name>` cut in two, refusing anything else.
modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go
+18 -18
View File
@@ -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)
}
modules/mesh-delivery-forge-bridge/internal/bridge/forgemade.go
+7 -6
View File
@@ -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
}
modules/mesh-delivery-forge-bridge/internal/bridge/moved.go
+95
View File
@@ -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 <owner>/<name> or branch (%q, %q)", e.Repository, e.Branch)
}
return m, nil
}
// Repository is `<owner>/<name>`.
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.<owner>.<name>.<escaped
// branch>` (A-236), apart from a pull request's `<owner>.<name>.<number>`.
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)
}
modules/mesh-delivery-forge-bridge/internal/bridge/moved_test.go
+77
View File
@@ -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)
}
}
}
modules/mesh-delivery-forge-bridge/internal/bridge/paths_test.go
+1 -1
View File
@@ -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)
}
modules/mesh-delivery-forge-bridge/internal/bridge/watched.go
+71
View File
@@ -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 `<owner>/<name>` 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
}
modules/mesh-delivery-forge-bridge/manifest_test.go
+5 -3
View File
@@ -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" {
modules/mesh-delivery-forge-bridge/module.json
+6
View File
@@ -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": [