FB2: the forge bridge says the finding branch-unprotected

A move of a watched branch whose protection was removed writes nothing and is said as the
finding branch-unprotected (scope merge, id the repository and branch as a key, warning) through
mesh-sdk's held finding writer into the bound state findings; a move that finds the protection
back clears it. The findings state is bound by the Binding with performed.

Builds: design 47 G §17, §131, §387, A-251
Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
2026-11-12 09:32:36 +01:00
parent ad561e485f
commit 103174e44c
7 changed files with 276 additions and 9 deletions
@@ -3,7 +3,9 @@ package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"sort"
"sync"
"github.com/nats-io/nats.go/jetstream"
@@ -91,6 +93,20 @@ func (s *boundBucket) Delete(ctx context.Context, key string, opts ...jetstream.
return kv.Delete(ctx, key, opts...)
}
// Keys are the bucket's keys, sorted; none is an empty list.
func (s *boundBucket) Keys(ctx context.Context) ([]string, error) {
kv, err := s.open(ctx)
if err != nil {
return nil, err
}
keys, err := kv.Keys(ctx)
if errors.Is(err, jetstream.ErrNoKeysFound) {
return []string{}, nil
}
sort.Strings(keys)
return keys, err
}
// keptCallOf is call as a kept call (ADR 0668 §1): asked again through a timeout or no responders until answered or
// busretry.Bound has passed.
func keptCallOf(ctx context.Context, call caller) func(key string, body any) (json.RawMessage, error) {
@@ -0,0 +1,111 @@
package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"time"
"github.com/nats-io/nats.go/jetstream"
condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/keys"
)
// The bridge's finding branch-unprotected (design 47 G §17, §387, §529): on every move of a watched branch the bridge reads the
// branch's protection again; a branch whose protection was removed starts nothing and is said as this finding, naming the
// branch, which the condition keeper raises and clears once the protection is back. The findings state is a `requires`
// entry opened by its binding (§577), written through mesh-sdk's held finding writer (§689).
const (
// findingsLocal is the local name of the state findings in the manifest's `requires`.
findingsLocal = "findings"
// kindBranchUnprotected is the finding's kind.
kindBranchUnprotected = "branch-unprotected"
// findingScope is the finding's scope (design 47 G A-251).
findingScope = "merge"
)
// findingWriter is the bridge's finding writer: mesh-sdk's held writer of the seat mesh-conditions' generated finding.
type findingWriter interface {
Write(ctx context.Context, f condseat.Finding) error
Clear(ctx context.Context, scope, id, kind, why string) error
Held() []condseat.Finding
}
// findingState is the state findings, a bound bucket, as the held writer's state.
type findingState struct{ b *boundBucket }
const findingStateTimeout = 10 * time.Second
func (s findingState) Get(key string) (*stdio.StateEntry, error) {
ctx, cancel := context.WithTimeout(context.Background(), findingStateTimeout)
defer cancel()
e, err := s.b.Get(ctx, key)
if errors.Is(err, jetstream.ErrKeyNotFound) {
return nil, nil
}
if err != nil {
return nil, err
}
return &stdio.StateEntry{Key: key, Value: json.RawMessage(e.Value()), Revision: e.Revision()}, nil
}
func (s findingState) Put(key string, value any) (uint64, error) {
data, err := json.Marshal(value)
if err != nil {
return 0, err
}
ctx, cancel := context.WithTimeout(context.Background(), findingStateTimeout)
defer cancel()
return s.b.Put(ctx, key, data)
}
func (s findingState) Delete(key string) error {
ctx, cancel := context.WithTimeout(context.Background(), findingStateTimeout)
defer cancel()
return s.b.Delete(ctx, key)
}
func (s findingState) Keys() ([]string, error) {
ctx, cancel := context.WithTimeout(context.Background(), findingStateTimeout)
defer cancel()
return s.b.Keys(ctx)
}
// unprotectedID is the finding's id: `<owner>/<repository>/<branch>` escaped as a key's part (A-251): a bus key holds no `~`, and EscapeName keeps `/`.
func unprotectedID(repository, branch string) string {
return keys.EscapeName(repository + "/" + branch)
}
// protection says what a read of a watched branch's protection means for its finding: a branch not protected is written as
// the finding branch-unprotected (the writer says it only when it is new or changed), a protected one clears it when it is held.
func (mv *mover) protection(ctx context.Context, repository, branch string, protected bool, head string) {
if mv.findings == nil {
return
}
id := unprotectedID(repository, branch)
if protected {
for _, f := range mv.findings.Held() {
if f.Scope == findingScope && f.ID == id && f.Kind == kindBranchUnprotected {
if err := mv.findings.Clear(ctx, findingScope, id, kindBranchUnprotected, "the protection of "+branch+" is back"); err != nil {
mv.logger.Printf("%s of %s: the finding %s is not cleared: %v", branch, repository, kindBranchUnprotected, err)
}
}
}
return
}
now := time.Now()
if mv.now != nil {
now = mv.now()
}
now = now.UTC()
err := mv.findings.Write(ctx, condseat.Finding{Scope: findingScope, ID: id, Kind: kindBranchUnprotected, Severity: condseat.FindingSeverityWarning,
Headline: "the branch " + branch + " of " + repository + " is not protected",
Summary: fmt.Sprintf("the branch %s of %s moved to %s while its protection is removed: no delivering pipeline starts for it until the protection is back", branch, repository, head),
Source: "git.branch", Observer: forgeBridgeModule, At: now, Subject: condseat.Subject{Scope: findingScope, ID: id}})
if err != nil {
mv.logger.Printf("%s of %s: the finding %s is not written: %v", branch, repository, kindBranchUnprotected, err)
}
}
@@ -66,6 +66,20 @@ accounts {
"$JS.API.STREAM.INFO.KV_mod_mesh-delivery-forge-bridge_performed_b1"
"$JS.FC.KV_mod_mesh-delivery-forge-bridge_performed_b1.>"
"$KV.mod_mesh-delivery-forge-bridge_performed_b1.>"
"$JS.API.CONSUMER.CREATE.KV_mod_mesh-delivery-forge-bridge_findings_b1.>"
"$JS.API.CONSUMER.DELETE.KV_mod_mesh-delivery-forge-bridge_findings_b1.>"
"$JS.API.DIRECT.GET.KV_mod_mesh-delivery-forge-bridge_findings_b1.>"
"$JS.API.STREAM.INFO.KV_mod_mesh-delivery-forge-bridge_findings_b1"
"$JS.FC.KV_mod_mesh-delivery-forge-bridge_findings_b1.>"
"$KV.mod_mesh-delivery-forge-bridge_findings_b1.>"
"$JS.API.STREAM.NAMES"
"$JS.API.CONSUMER.CREATE.EVENTS"
"$JS.API.CONSUMER.CREATE.EVENTS.>"
"$JS.API.CONSUMER.DELETE.EVENTS.>"
"$JS.API.CONSUMER.MSG.NEXT.EVENTS.>"
"mesh.stream.mesh-delivery-forge-bridge.event.finding-written.n1"
"mesh.stream.mesh-delivery-forge-bridge.event.finding-cleared.n1"
"mesh.stream.mesh-delivery-forge-bridge.event.findings-stated.n1"
"$JS.API.CONSUMER.CREATE.KV_mesh-delivery-forge-bridge_inputs.>"
"$JS.API.CONSUMER.DELETE.KV_mesh-delivery-forge-bridge_inputs.>"
"$JS.API.DIRECT.GET.KV_mesh-delivery-forge-bridge_inputs.>"
@@ -217,6 +231,9 @@ func fileAnswer(path, ref, content string, found bool) string {
// boundPerformed is the bus name the stand-in seat mesh-broker's show answers for the state performed.
const boundPerformed = "mod_mesh-delivery-forge-bridge_performed_b1"
// boundFindings is the bus name the stand-in seat mesh-broker's show answers for the state findings.
const boundFindings = "mod_mesh-delivery-forge-bridge_findings_b1"
// startBed is startBridge with a step between the streams and the consumers: prepare writes what the streams hold before
// the bridge's consumers are made, as the grants part makes them, from now (ADR 0443 §1), and reply sets the answer of the
// seat git's open-pulls from then on. A stand-in holds the seats the bridge calls: git (every verb, trees r0 by default,
@@ -237,7 +254,12 @@ func startBed(t *testing.T, openReply string, prepare func(ctx context.Context,
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.*.event.>"}}); err != nil {
t.Fatal(err)
}
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "FORGE", Subjects: []string{"mesh.stream.mesh-delivery-forge-bridge.event.>"}}); err != nil {
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "FORGE", Subjects: []string{"mesh.stream.mesh-delivery-forge-bridge.event.forge.>", "mesh.stream.mesh-delivery-forge-bridge.event.owed.ended"}}); err != nil {
t.Fatal(err)
}
// The bridge's own stream of its findings' events (design 47 G §383, §529), which its finding writer says into.
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "mesh-delivery-forge-bridge-findings", Subjects: []string{"mesh.stream.mesh-delivery-forge-bridge.event.finding-written.*",
"mesh.stream.mesh-delivery-forge-bridge.event.finding-cleared.*", "mesh.stream.mesh-delivery-forge-bridge.event.findings-stated.*"}}); err != nil {
t.Fatal(err)
}
// Each seat's own stream owed, as its contract keeps it (Stream<E>, SEAT_<seat>_owed), as the grants part makes it: no
@@ -250,7 +272,7 @@ func startBed(t *testing.T, openReply string, prepare func(ctx context.Context,
}
// The bridge's states, made from the manifest's state entries and its requires before the process starts; the state
// performed under the bus name the seat mesh-broker's show answers for it (boundPerformed), never an old name (§577).
for _, name := range []string{headsBucket, boundPerformed, inputsBucket, groupkeys.EndAsksBucket, groupkeys.OwedAsksBucket} {
for _, name := range []string{headsBucket, boundPerformed, boundFindings, inputsBucket, groupkeys.EndAsksBucket, groupkeys.OwedAsksBucket} {
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: 1}); err != nil {
t.Fatal(err)
}
@@ -279,8 +301,9 @@ func startBed(t *testing.T, openReply string, prepare func(ctx context.Context,
"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 + `"}}`,
"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 + `"}}`,
"mesh-broker.show findings": `{"state":"created","binding":{"local":"findings","storage":"key-value","template":"state","bucket":"` + boundFindings + `"}}`,
}
for _, verb := range []string{"status", "comment", "view", "note", "merge-pull"} {
answers["git."+verb] = string(gitfixture.Given(t, verb+" answers", "verb:"+verb+".answer"))
@@ -294,10 +317,16 @@ 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 }
var in struct {
Repository string
Local string `json:"local-name"`
}
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)
}
if in.Local != "" && answers[verb+" "+in.Local] != "" {
answer = answers[verb+" "+in.Local] // show answers each entry of the module's requires by its local name
}
mu.Unlock()
if verb == "git.file" {
var in git.FileInput
@@ -873,3 +902,35 @@ func TestBridgePerformsWhatIsOwedOnce(t *testing.T) {
}
func sha256Sum(b []byte) []byte { s := sha256.Sum256(b); return s[:] }
// Design 47 G §17, §383, §529, A-251 (FB2): the bridge started whole against the real bus says a move of a branch whose protection
// was removed as the finding branch-unprotected, into its bound state findings and its finding stream, and writes no message.
func TestBridgeSaysAnUnprotectedBranchAsAFinding(t *testing.T) {
b := startBed(t, `{"pulls":[]}`, nil)
b.answer("git.branch novox/mesh-delivery", `{"repository":"novox/mesh-delivery","branch":"main","head":"`+merge+`","protected":false}`)
moveMain(t, b, "novox/mesh-delivery")
kv, err := b.js.KeyValue(b.ctx, boundFindings)
if err != nil {
t.Fatal(err)
}
var e jetstream.KeyValueEntry
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
if e, err = kv.Get(b.ctx, "merge.novox/mesh-delivery/main.branch-unprotected"); err == nil {
break
}
}
if err != nil {
t.Fatalf("the finding is not in the bound state findings: %v", err)
}
var f struct{ Kind, Scope, ID, Observer string }
if err := json.Unmarshal(e.Value(), &f); err != nil || f.Kind != "branch-unprotected" || f.Observer != "mesh-delivery-forge-bridge" {
t.Errorf("the finding = %s %v", e.Value(), err)
}
forge, err := b.js.Stream(b.ctx, "FORGE")
if err != nil {
t.Fatal(err)
}
if info, err := forge.Info(b.ctx); err != nil || info.State.Msgs != 1 { // the open message of its start only
t.Errorf("FORGE holds %+v %v: an unprotected branch's move writes no message", info, err)
}
}
@@ -23,6 +23,7 @@ import (
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
"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"
@@ -147,15 +148,29 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
// The state performed is a `requires` entry: opened by its binding, asked of the seat mesh-broker's show at this start
// as a kept call and again on its recreated and dropped (design 47 G §577). Until show answers, nothing owed is
// performed and the bridge says it waits, naming the entry.
binding := process.NewBinding(brokerSeat, forgeBridgeModule, keptCallOf(ctx, call), logger.Printf, performedLocal)
binding := process.NewBinding(brokerSeat, forgeBridgeModule, keptCallOf(ctx, call), logger.Printf, performedLocal, findingsLocal)
if err := binding.Register(hub); err != nil {
return err
}
binding.Ask(performedLocal)
binding.Ask(findingsLocal)
if w := binding.Waiting(); w != "" {
logger.Printf("the bridge %s: nothing owed to the forge is performed until it is bound", w)
}
performedKV := &boundBucket{binding: binding, local: performedLocal, js: js}
// The bridge's one finding, branch-unprotected (design 47 G §17, §387): written through mesh-sdk's held finding writer into the
// state findings, a `requires` entry opened by its binding; a writer that cannot start (the entry is not bound, or the stream
// of its asks is not there yet) is said, and what it holds is stated at the next start.
findingsWriter, err := views.NewHeldWriter(views.WriterConfig[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent]{
Module: cred.Module, State: findingState{&boundBucket{binding: binding, local: findingsLocal, js: js}}, Node: cred.Node,
Asked: condseat.SubjectFindingsAsked, JS: js, Conn: conn, Logf: logger.Printf})
if err != nil {
return fmt.Errorf("the bridge's finding writer cannot be made: %w", err)
}
defer findingsWriter.Close()
if err := findingsWriter.Start(ctx); err != nil {
logger.Printf("the finding writer cannot start, the finding branch-unprotected is stated at the next start: %v", err)
}
// What the gate and the checker owe the forge, performed through the seat git (design 47 *What the gate owes the forge*).
owe := &owing{call: call, pub: js, performed: kvPerformed{performedKV}, waits: binding.Waiting, now: time.Now, logger: logger}
// A delivery group's member names its group, and its end is said once (ADR 0633 §4), from the seat mesh-delivery's groups.
@@ -199,7 +214,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
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}
mv = &mover{call: call, pub: rec, heads: heads, watched: watched, inputs: inputs, groups: groups, findings: findingsWriter, 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.
@@ -13,6 +13,7 @@ import (
"testing"
"time"
condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
git "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/git"
"git.novox.be/novox/mesh-sdk/go/process"
@@ -297,3 +298,59 @@ func someFiles(_, _, path, _ string) ([]byte, error) {
type refusingBus struct{ err error }
func (r refusingBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) { return nil, r.err }
// fakeFindings is a finding writer that holds what it is given.
type fakeFindings struct {
held []condseat.Finding
cleared []string
}
func (f *fakeFindings) Write(_ context.Context, w condseat.Finding) error {
f.held = append(f.held, w)
return nil
}
func (f *fakeFindings) Clear(_ context.Context, scope, id, kind, _ string) error {
f.cleared = append(f.cleared, scope+"."+id+"."+kind)
f.held = nil
return nil
}
func (f *fakeFindings) Held() []condseat.Finding { return f.held }
// Design 47 G §17, A-251 (FB2): a move of a watched branch whose protection was removed writes nothing and says the finding
// branch-unprotected, scope merge, its id the owner, repository and branch escaped as a key; a move that finds the protection
// back clears it, and a protected branch with no finding held says nothing.
func TestAnUnprotectedBranchIsSaidAsAFindingAndClearedWhenItIsBack(t *testing.T) {
var logs bytes.Buffer
protected := false
call := func(_ context.Context, key string, _ any) (json.RawMessage, error) {
if key == "seat:git.branch" {
return json.RawMessage(fmt.Sprintf(`{"repository":"novox/mesh-delivery","branch":"beta","head":"%s","protected":%t}`, landed, protected)), nil
}
return nil, errors.New("not asked: " + key)
}
pub, heads, findings := &fakePub{}, newHeads(), &fakeFindings{}
mv := newMover(call, pub, heads, &logs)
mv.findings, mv.now = findings, func() time.Time { return time.Date(2026, 11, 12, 9, 0, 0, 0, time.UTC) }
mv.watched = &bridge.Watched{}
mv.watched.Set(nil, []string{"beta"})
mv.hear(context.Background(), &fakeMsg{subject: bridge.BranchMovedSubject, data: movedEvent("beta")})
if len(pub.subjects) != 0 || len(findings.held) != 1 {
t.Fatalf("written %v, findings %+v: want nothing written and one finding", pub.subjects, findings.held)
}
f := findings.held[0]
if f.Scope != "merge" || f.ID != "novox/mesh-delivery/beta" || f.Kind != "branch-unprotected" || f.Severity != "warning" || f.Observer != "mesh-delivery-forge-bridge" ||
f.Subject.Scope != "merge" || f.Subject.ID != "novox/mesh-delivery/beta" || !strings.Contains(f.Summary, "beta") {
t.Errorf("the finding = %+v", f)
}
// The protection is back at a move whose head the bridge wrote already: the finding is cleared.
protected = true
heads.vals["branch.novox.mesh-delivery.beta"] = landed
mv.hear(context.Background(), &fakeMsg{subject: bridge.BranchMovedSubject, data: movedEvent("beta")})
if len(findings.cleared) != 1 || findings.cleared[0] != "merge.novox/mesh-delivery/beta.branch-unprotected" || len(findings.held) != 0 {
t.Errorf("cleared %v held %+v, want the finding cleared", findings.cleared, findings.held)
}
mv.hear(context.Background(), &fakeMsg{subject: bridge.BranchMovedSubject, data: movedEvent("beta")})
if len(findings.cleared) != 1 {
t.Errorf("cleared %v: a protected branch with no finding held clears nothing", findings.cleared)
}
}
@@ -7,6 +7,7 @@ import (
"log"
"strings"
"sync"
"time"
"github.com/nats-io/nats.go/jetstream"
@@ -28,8 +29,11 @@ type mover struct {
watched *bridge.Watched
inputs bridge.Kept
groups *grouper
turn *sync.Mutex
logger *log.Logger
// findings says the finding branch-unprotected (design 47 G §17, A-251); nil says none.
findings findingWriter
now func() time.Time
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
@@ -66,6 +70,8 @@ func (mv *mover) move(ctx context.Context, moved bridge.Moved) error {
if err != nil {
return fmt.Errorf("%s of %s: %w", moved.Branch, moved.Repository(), err)
}
// The protection is read at every move, so a finding stands until a move finds it back (§17).
mv.protection(ctx, moved.Repository(), moved.Branch, b.Protected, b.Head)
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)
@@ -4,6 +4,7 @@ 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-contracts/health-and-repair/go/seats/meshconditions 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