D1 (forge bridge): the bridge names its own module from its credential's
Its states' buckets, its stream's subjects and its finding observer take the module its credential names (the tools: the module the runtime serves them as) instead of constants of its name; a guard test finds none in its code. Builds: design 47 G A-296, §428 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -181,8 +181,8 @@ func read(of []seat.Group, performed, asks stdio.KeptState) ([]groups.Row, []str
|
||||
}
|
||||
|
||||
// askEnd writes the end-asks value at key and waits at most bound for the bridge's answer.
|
||||
func askEnd(s stdio.KeptState, bound time.Duration) asker {
|
||||
wait := askState(s, groups.EndAsksBucket, bound, func(v []byte) bool { a, err := groups.DecodeEndAsk(v); return err == nil && a.Done() })
|
||||
func askEnd(s stdio.KeptState, module string, bound time.Duration) asker {
|
||||
wait := askState(s, groups.EndAsksBucket(module), bound, func(v []byte) bool { a, err := groups.DecodeEndAsk(v); return err == nil && a.Done() })
|
||||
return func(key string, value []byte) (groups.EndAsk, error) {
|
||||
v, err := wait(key, value)
|
||||
if err != nil {
|
||||
@@ -246,8 +246,10 @@ func askCaller(_ context.Context, key string, body any) (json.RawMessage, error)
|
||||
}
|
||||
|
||||
func main() {
|
||||
// The module the runtime serves these tools as: the bridge's, whose states they ask (design 47 G A-296).
|
||||
served := os.Getenv(stdio.ServedAs)
|
||||
performed, asks := stdio.State(groups.PerformedState), stdio.State(groups.EndAsksState)
|
||||
tools := Tools(askEnd(asks, 25*time.Second), time.Now,
|
||||
tools := Tools(askEnd(asks, served, 25*time.Second), time.Now,
|
||||
func() ([]groups.Row, []string, error) {
|
||||
of, err := groups.Read(context.Background(), askCaller)
|
||||
if err != nil {
|
||||
@@ -265,7 +267,7 @@ func main() {
|
||||
}, askCaller)
|
||||
owedAsks := stdio.State(groups.OwedAsksState)
|
||||
recorder := meshhandactsclient.Recorder{Ask: stdio.Ask}
|
||||
tools = append(tools, OwedTools(askState(owedAsks, groups.OwedAsksBucket, 25*time.Second, func(v []byte) bool {
|
||||
tools = append(tools, OwedTools(askState(owedAsks, groups.OwedAsksBucket(served), 25*time.Second, func(v []byte) bool {
|
||||
a, err := groups.DecodeOwedAsk(v)
|
||||
return err == nil && a.Done()
|
||||
}), func() ([]OwedRow, error) { return performedRows(performed) }, func(what, why string) error {
|
||||
|
||||
@@ -37,7 +37,7 @@ func performedIsOpenedByItsBinding(t *testing.T) {
|
||||
}
|
||||
return json.RawMessage(`{"state":"created","binding":{"local":"performed","storage":"key-value","bucket":"bound-performed"}}`), nil
|
||||
}
|
||||
binding := process.NewBinding(brokerSeat, forgeBridgeModule, show, nil, performedLocal)
|
||||
binding := process.NewBinding(brokerSeat, forgeBridgeModule(), show, nil, performedLocal)
|
||||
bus := &openerFake{}
|
||||
kv := &boundBucket{binding: binding, local: performedLocal, js: bus}
|
||||
binding.Ask(performedLocal)
|
||||
|
||||
@@ -104,7 +104,7 @@ func (mv *mover) protection(ctx context.Context, repository, branch string, prot
|
||||
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}})
|
||||
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)
|
||||
}
|
||||
|
||||
+3
-3
@@ -279,7 +279,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(forgeBridgeModule), boundPerformed, boundFindings, inputsBucket(forgeBridgeModule), groupkeys.EndAsksBucket, groupkeys.OwedAsksBucket} {
|
||||
for _, name := range []string{headsBucket(forgeBridgeModule()), boundPerformed, boundFindings, inputsBucket(forgeBridgeModule()), groupkeys.EndAsksBucket(forgeBridgeModule()), groupkeys.OwedAsksBucket(forgeBridgeModule())} {
|
||||
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: 1}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -651,7 +651,7 @@ func TestBridgeReadsTheForgeAtStartAndAfterAReconnectAndHearsNoOlderEvent(t *tes
|
||||
}
|
||||
}
|
||||
// What the bridge kept of an earlier write: the head of mesh-delivery#12.
|
||||
kv, err := js.KeyValue(ctx, headsBucket(forgeBridgeModule))
|
||||
kv, err := js.KeyValue(ctx, headsBucket(forgeBridgeModule()))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -877,7 +877,7 @@ func TestBridgePerformsWhatIsOwedOnce(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
asks, err := b.js.KeyValue(b.ctx, groupkeys.OwedAsksBucket)
|
||||
asks, err := b.js.KeyValue(b.ctx, groupkeys.OwedAsksBucket(forgeBridgeModule()))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
@@ -42,7 +42,7 @@ const eventsStream = "EVENTS"
|
||||
// The bridge's own stream FORGE takes `mesh.stream.<module>.event.<event>` (mesh-control internal/broker
|
||||
// modstreams.go ModuleStreamSubject).
|
||||
func forgeSubject(event string) string {
|
||||
return process.ModuleStreamSubject(forgeBridgeModule, event)
|
||||
return process.ModuleStreamSubject(forgeBridgeModule(), event)
|
||||
}
|
||||
|
||||
func main() {
|
||||
@@ -114,6 +114,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
setForgeBridgeModule(cred.Module) // the bridge names its own states and stream from its credential's module (A-296, §428)
|
||||
again := make(chan struct{}, 1)
|
||||
conn, js, err := cred.Dial(
|
||||
nats.ReconnectHandler(func(*nats.Conn) {
|
||||
@@ -148,7 +149,7 @@ 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, findingsLocal, views.FindingEventsLocal)
|
||||
binding := process.NewBinding(brokerSeat, forgeBridgeModule(), keptCallOf(ctx, call), logger.Printf, performedLocal, findingsLocal, views.FindingEventsLocal)
|
||||
if err := binding.Register(hub); err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -180,7 +181,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
groups := &grouper{groups: groupsVia(call), send: sendVia(js), done: kvHeads{performedKV}, logger: logger}
|
||||
// A person says a group's end again with bridge_group_end_again, which writes the state end-asks (ADR 0633 §4): watched, and
|
||||
// answered there; a state not there at this start is answered from the next start.
|
||||
if kv := process.OptionalBucket(ctx, js, groupkeys.EndAsksBucket, "bridge_group_end_again is not answered until the bridge starts again", logger.Printf); kv != nil {
|
||||
if kv := process.OptionalBucket(ctx, js, groupkeys.EndAsksBucket(forgeBridgeModule()), "bridge_group_end_again is not answered until the bridge starts again", logger.Printf); kv != nil {
|
||||
go func() {
|
||||
err := watch(ctx, kv, func(key string, rev uint64, value []byte) { groups.endAsked(ctx, kv, key, rev, value) })
|
||||
if err != nil && ctx.Err() == nil {
|
||||
@@ -189,7 +190,7 @@ 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}
|
||||
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
|
||||
@@ -290,7 +291,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
}
|
||||
// A person performs an act again, drops one or reads the consumers' backlog with the tools bridge_owed_again,
|
||||
// bridge_owed_drop and bridge_owed, which write the state owed-asks: watched, and answered there.
|
||||
if kv := process.OptionalBucket(ctx, js, groupkeys.OwedAsksBucket, "bridge_owed_again, bridge_owed_drop and bridge_owed's backlog are not answered until the bridge starts again", logger.Printf); kv != nil {
|
||||
if kv := process.OptionalBucket(ctx, js, groupkeys.OwedAsksBucket(forgeBridgeModule()), "bridge_owed_again, bridge_owed_drop and bridge_owed's backlog are not answered until the bridge starts again", logger.Printf); kv != nil {
|
||||
backlog := owedBacklog(js, cred, time.Now)
|
||||
go func() {
|
||||
err := watch(ctx, kv, func(key string, rev uint64, value []byte) {
|
||||
|
||||
@@ -1,10 +1,16 @@
|
||||
package main
|
||||
|
||||
import groupkeys "mesh-delivery-forge-bridge/internal/groups"
|
||||
import "sync/atomic"
|
||||
|
||||
// The holders and buckets the forge bridge names, taken from its own manifest (module.json: its name and its
|
||||
// states); a unit row checks them against it (TestTheNamesAreTheManifests). No contract names a holder
|
||||
// (novox/hq ADR 0625 §4, ADR 0634 §1).
|
||||
const (
|
||||
forgeBridgeModule = groupkeys.Module
|
||||
)
|
||||
// The module the bridge runs as: its credential's at its start (the tools program's: the module the runtime serves it as),
|
||||
// never a constant of its name (design 47 G A-296, §428). Its manifest's name is held equal by TestTheNamesAreTheManifests.
|
||||
var ownModule atomic.Value
|
||||
|
||||
// setForgeBridgeModule says the module the bridge runs as, once at its start.
|
||||
func setForgeBridgeModule(module string) { ownModule.Store(module) }
|
||||
|
||||
// forgeBridgeModule is the module the bridge runs as.
|
||||
func forgeBridgeModule() string {
|
||||
m, _ := ownModule.Load().(string)
|
||||
return m
|
||||
}
|
||||
|
||||
@@ -25,15 +25,15 @@ func theNamesAreTheManifests(t *testing.T) {
|
||||
if err := json.Unmarshal(raw, &m); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if m.Module != forgeBridgeModule {
|
||||
t.Errorf("the manifest's module is %q, the bridge names itself %q", m.Module, forgeBridgeModule)
|
||||
if m.Module != forgeBridgeModule() {
|
||||
t.Errorf("the manifest's module is %q, the bridge names itself %q", m.Module, forgeBridgeModule())
|
||||
}
|
||||
if len(m.Reads) != 0 {
|
||||
t.Errorf("the bridge reads no other module's state (acceptance 49), its manifest's reads are %v", m.Reads)
|
||||
}
|
||||
var own []string
|
||||
for _, s := range m.State {
|
||||
own = append(own, forgeBridgeModule+"_"+s.Name)
|
||||
own = append(own, forgeBridgeModule()+"_"+s.Name)
|
||||
}
|
||||
for _, r := range m.Requires {
|
||||
var kept struct {
|
||||
@@ -41,14 +41,14 @@ func theNamesAreTheManifests(t *testing.T) {
|
||||
Template string `json:"template"`
|
||||
}
|
||||
if json.Unmarshal(r, &kept) == nil && kept.Template == "state" {
|
||||
own = append(own, forgeBridgeModule+"_"+kept.Name)
|
||||
own = append(own, forgeBridgeModule()+"_"+kept.Name)
|
||||
}
|
||||
}
|
||||
// Design 47 G A-296: its own buckets are named from its credential's module, never a constant.
|
||||
if headsBucket("n1-bridge") != "n1-bridge_heads" || inputsBucket("n1-bridge") != "n1-bridge_inputs" {
|
||||
t.Errorf("the own buckets of n1-bridge are %s and %s", headsBucket("n1-bridge"), inputsBucket("n1-bridge"))
|
||||
}
|
||||
for _, b := range []string{headsBucket(forgeBridgeModule), inputsBucket(forgeBridgeModule)} {
|
||||
for _, b := range []string{headsBucket(forgeBridgeModule()), inputsBucket(forgeBridgeModule())} {
|
||||
if !slices.Contains(own, b) {
|
||||
t.Errorf("the bridge keeps %s, its manifest's states are %v", b, own)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,37 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"io/fs"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// The tests run the bridge as the module its manifest names, as its credential would say it.
|
||||
func init() { setForgeBridgeModule("mesh-delivery-forge-bridge") }
|
||||
|
||||
// No code of the bridge names its own module: its buckets, its stream's subjects and its finding observer take the module its
|
||||
// credential (or the runtime that serves its tools) names (design 47 G A-296, §428). The manifest alone says it.
|
||||
func TestNoCodeNamesTheBridgesOwnModule(t *testing.T) {
|
||||
name := regexp.MustCompile(`"mesh-delivery-forge-bridge(_[^"]*)?"`)
|
||||
err := filepath.WalkDir("../..", func(path string, d fs.DirEntry, err error) error {
|
||||
if err != nil || d.IsDir() || !strings.HasSuffix(path, ".go") || strings.HasSuffix(path, "_test.go") {
|
||||
return err
|
||||
}
|
||||
src, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for i, line := range strings.Split(string(src), "\n") {
|
||||
if !strings.HasPrefix(strings.TrimSpace(line), "//") && name.MatchString(line) {
|
||||
t.Errorf("%s:%d names the bridge's own module: %s", path, i+1, strings.TrimSpace(line))
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
@@ -24,16 +24,15 @@ import (
|
||||
|
||||
// The bridge's states the tools read and write.
|
||||
const (
|
||||
// Module is the forge bridge's own name (its manifest's module, checked by the bridge's names test).
|
||||
Module = "mesh-delivery-forge-bridge"
|
||||
// PerformedState is the bridge's state of what it performed and published.
|
||||
PerformedState = "performed"
|
||||
// EndAsksState is the state of asks to say a group's end again, keyed by the group escaped as a key.
|
||||
EndAsksState = "end-asks"
|
||||
// EndAsksBucket is its bucket on the bus.
|
||||
EndAsksBucket = Module + "_" + EndAsksState
|
||||
)
|
||||
|
||||
// EndAsksBucket is its bucket on the bus, `<module>_<state>` of the module the bridge runs as, its credential's (A-296).
|
||||
func EndAsksBucket(module string) string { return module + "_" + EndAsksState }
|
||||
|
||||
// Key is the key of performed that says a group's end on a base is written: `groups.<group>.<base>`, each part escaped as
|
||||
// a key name, its value the last member's merge commit the end named (design 47 G §715: one end per base).
|
||||
func Key(group, base string) (string, error) {
|
||||
|
||||
@@ -15,10 +15,11 @@ import (
|
||||
const (
|
||||
// OwedAsksState is the state of asks about what is owed, one key per kind of ask and act.
|
||||
OwedAsksState = "owed-asks"
|
||||
// OwedAsksBucket is its bucket on the bus.
|
||||
OwedAsksBucket = Module + "_" + OwedAsksState
|
||||
)
|
||||
|
||||
// OwedAsksBucket is its bucket on the bus, `<module>_<state>` of the module the bridge runs as, its credential's (A-296).
|
||||
func OwedAsksBucket(module string) string { return module + "_" + OwedAsksState }
|
||||
|
||||
// The kinds of an owed ask.
|
||||
const (
|
||||
OwedAgain = "again"
|
||||
|
||||
Reference in New Issue
Block a user