Compare commits

..
15 changed files with 117 additions and 1069 deletions
@@ -1,78 +0,0 @@
package main
import (
"context"
"encoding/json"
"testing"
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/nats-io/nats.go/jetstream"
)
// raceKV is a digests bucket in which another writer writes its composition and digest just before this writer reads
// the digest key: what two take-ins composing one node at once do.
type raceKV struct {
*fakeComposedKV
before func()
}
func (r *raceKV) Get(ctx context.Context, key string) (jetstream.KeyValueEntry, error) {
if r.before != nil {
b := r.before
r.before = nil
b()
}
return r.fakeComposedKV.Get(ctx, key)
}
// The digest kept beside a composition is always the digest of the composition the bucket holds: when another writer
// writes its composition and its digest between this writer's composition and its digest, this writer's digest must not
// win, or the assigner's want names a composition the bucket no longer holds and the node applies nothing (issue 1945,
// live on 2026-10-25: shanks' node-tools composition had digest 575a5c96, its want named a36e63c8).
func TestTheDigestKeptIsTheDigestOfTheCompositionHeld(t *testing.T) {
at, _, _, _ := aPlace(t, "shanks")
key, _ := sdkdelivery.ComposedKey("shanks", "node-tools", fp2Of('a'))
file := func(content string) json.RawMessage {
return json.RawMessage(`{"id":"node-tools.f","type":"file","module":"node-tools","path":"/etc/nt","mode":"0644","content":"` +
content + `"}`)
}
ours := sdkdelivery.Composed{Node: "shanks", Module: "node-tools", FP: fp2Of('a'), Resources: []json.RawMessage{file("ours")}}
theirs := sdkdelivery.Composed{Node: "shanks", Module: "node-tools", FP: fp2Of('a'), Resources: []json.RawMessage{file("theirs")}}
race := &raceKV{fakeComposedKV: at.digests.(*fakeComposedKV)}
at.digests = race
race.before = func() {
// The other writer: its composition over ours, then its digest.
other := at
other.digests = race.fakeComposedKV
if _, err := writeComposed(context.Background(), other, key, theirs); err != nil {
t.Fatal(err)
}
}
if _, err := writeComposed(context.Background(), at, key, ours); err != nil {
t.Fatal(err)
}
held, _, err := readComposed(context.Background(), at.bucket, key)
if err != nil {
t.Fatal(err)
}
heldDigest, _ := sdkdelivery.ComposedDigestOf(held)
e, err := race.fakeComposedKV.Get(context.Background(), key)
if err != nil {
t.Fatal(err)
}
kept, err := sdkdelivery.DecodeComposedDigest(e.Value())
if err != nil {
t.Fatal(err)
}
if kept.Digest != heldDigest {
t.Fatalf("the digest kept is %s and the composition held has %s", kept.Digest, heldDigest)
}
}
func fp2Of(c byte) string {
b := make([]byte, 64)
for i := range b {
b[i] = c
}
return sdkdelivery.SourcePrefix2 + string(b)
}
+21 -96
View File
@@ -120,13 +120,6 @@ var (
doneKeysSay = func(s string) { fmt.Println(s) }
)
// doneKeysBeforeLetGo is called just before a follower lets go of its module; a variable so a test can wake the module
// at that moment. Nil does nothing.
var doneKeysBeforeLetGo func(module string)
// doneKeysLookAgain marks a module woken while a take-in of it ran: that take-in goes round once more.
var doneKeysLookAgain sync.Map
// doneKeysMu keeps one take-in of a module's done key at a time in this process, per module.
var doneKeysMu sync.Map
@@ -136,42 +129,18 @@ var doneKeysMu sync.Map
// yet is no failure: it is said once for each reason it waits for and looked at again every doneKeyRetry.
func takeDoneKey(ctx context.Context, open *stores, module string) {
mu, _ := doneKeysMu.LoadOrStore(module, &sync.Mutex{})
for {
if !mu.(*sync.Mutex).TryLock() {
// A take-in of the module already follows it: it is told to look again once it is done, so a newer key
// written meanwhile is never lost (the review of issue 1945).
doneKeysLookAgain.Store(module, true)
return
}
followDoneKey(ctx, open, module)
if doneKeysBeforeLetGo != nil {
doneKeysBeforeLetGo(module)
}
mu.(*sync.Mutex).Unlock()
// A wake between this follower's last look and its letting go failed its TryLock and only marked the module:
// looked at again now that it is let go, and taken once more (the review of issue 1945).
if _, again := doneKeysLookAgain.LoadAndDelete(module); !again || ctx.Err() != nil {
return
}
if !mu.(*sync.Mutex).TryLock() {
return // a take-in of the module already follows it, and looks again within doneKeyRetry
}
}
// followDoneKey takes a module's done key in, under the module's lock, until it is taken in or ctx ends: a failure or a
// wait is tried again every doneKeyRetry.
func followDoneKey(ctx context.Context, open *stores, module string) {
defer mu.(*sync.Mutex).Unlock()
failed := false
saidWait := ""
for {
doneKeysLookAgain.Delete(module) // what this look reads covers every wake before it
err := doneKeyTakeOnce(ctx, open, module)
o := doneKeyObservation(module)
if err == nil {
if keeper := conditionsFrom; keeper != nil && failed {
_, _ = keeper.Clear(ctx, o.Key(), "composed and granted")
failed = false
}
if _, again := doneKeysLookAgain.LoadAndDelete(module); again {
continue // a wake came while this one ran: once more, for the key written meanwhile
}
return
}
@@ -282,13 +251,21 @@ func liveDoneKeySteps(open *stores, keys registerKeys) doneKeySteps {
}
return holders[catalogue.BrokerSeat].Node, nil
},
// The composer's, never the take-in's own: one pass at a time composes each node asked for once, and writes the
// grants once (issue 1945, composer.go).
compose: func(ctx context.Context, nodes []string) error {
return composerFor(open).compose(ctx, nodes, false)
_, err := composeModules(ctx, open, nodes)
return err
},
grants: func(ctx context.Context) error {
return composerFor(open).compose(ctx, nil, true)
bus, err := aBus()
if err != nil {
return err
}
defer bus.Close()
js, err := jetstream.New(bus.Conn())
if err != nil {
return err
}
return grantsIssued(ctx, inv, js, "done keys: ")
},
}
}
@@ -330,44 +307,23 @@ func doneKeysAtStart(ctx context.Context, modules func(context.Context) ([]strin
return
}
sort.Strings(names)
// Side by side, a few at a time: their compositions meet in the composer's passes, each node composed once per pass
// (issue 1945), so the start's look is not one module's composition after another.
sem := make(chan struct{}, startTakeIns)
var wg sync.WaitGroup
for _, m := range names {
if ctx.Err() != nil {
break
return
}
sem <- struct{}{}
wg.Add(1)
go func() {
defer func() { <-sem; wg.Done() }()
take(m)
}()
take(m)
}
wg.Wait()
}
// startTakeIns is how many take-ins the start's look runs at once.
const startTakeIns = 16
// takeDoneKeyAtStart is one take-in of the start's look, under the module's lock; one that fails is handed to takeDoneKey,
// which tries again every doneKeyRetry, loudly, so a failing module never holds up the look at the others.
func takeDoneKeyAtStart(ctx context.Context, open *stores, module string) {
mu, _ := doneKeysMu.LoadOrStore(module, &sync.Mutex{})
if !mu.(*sync.Mutex).TryLock() {
// A take-in already follows the module: told to look again, it takes what the start would have (issue 1945).
doneKeysLookAgain.Store(module, true)
return
}
doneKeysLookAgain.Delete(module) // what this look reads covers every wake before it
err := doneKeyTakeOnce(ctx, open, module)
mu.(*sync.Mutex).Lock()
err := takeDoneKeyOnce(ctx, open, module)
mu.(*sync.Mutex).Unlock()
// Read after letting go, as every holder of a module's lock does: a wake during this look only marked the module
// (the review of issue 1945).
_, again := doneKeysLookAgain.LoadAndDelete(module)
if err != nil || again {
// A failure, a wait or a wake meanwhile is followed by takeDoneKey, which says it and looks again.
if err != nil {
// A failure or a wait is followed by takeDoneKey, which says it and looks again (issue 1945).
inBackground(func() { takeDoneKey(ctx, open, module) })
}
}
@@ -386,34 +342,3 @@ func catalogedModules(open *stores) func(context.Context) ([]string, error) {
return out, nil
}
}
// doneKeysComposers are the composers of the done keys' take-ins in this process, one for each set of stores (a test
// runs several meshes), each started at its first use and run for as long as the process serves.
var doneKeysComposers sync.Map
// composerFor is the composer of the stores, its passes composing on them and writing the grants on the bus.
func composerFor(open *stores) *composer {
if c, ok := doneKeysComposers.Load(open); ok {
return c.(*composer)
}
c := newComposer(func(ctx context.Context, nodes []string) map[string]error {
_, failed := composeModulesByNode(ctx, open, nodes)
return failed
}, func(ctx context.Context) error {
bus, err := aBus()
if err != nil {
return err
}
defer bus.Close()
js, err := jetstream.New(bus.Conn())
if err != nil {
return err
}
return grantsIssued(ctx, open.inventory, js, "done keys: ")
})
if was, loaded := doneKeysComposers.LoadOrStore(open, c); loaded {
return was.(*composer)
}
go c.run(context.Background())
return c
}
@@ -5,7 +5,6 @@ import (
"errors"
"reflect"
"sort"
"sync"
"testing"
)
@@ -16,10 +15,9 @@ import (
// new fingerprints, and the new one, watching updates only, never looked).
func TestAStartingControllerTakesEveryModulesDoneKeyOnce(t *testing.T) {
var took []string
var mu sync.Mutex
doneKeysAtStart(context.Background(), func(context.Context) ([]string, error) {
return []string{"mesh-provisioner-client", "gitea", "postgres"}, nil
}, func(module string) { mu.Lock(); took = append(took, module); mu.Unlock() })
}, func(module string) { took = append(took, module) })
sort.Strings(took)
if want := []string{"gitea", "mesh-provisioner-client", "postgres"}; !reflect.DeepEqual(took, want) {
t.Fatalf("took %v, want %v", took, want)
+1 -1
View File
@@ -99,7 +99,7 @@ func TestAChangedUserListIsANewRevisionOfTheBusModulesCompositionAlone(t *testin
d["resources"] = append(d["resources"].([]any), map[string]any{"id": "nats.bus-users", "type": "file",
"module": "nats", "path": "/var/lib/mesh/nats/users.conf", "content": users, "mode": "0600"})
raw, _ := json.Marshal(d)
comps, _, _, err := compositionsOf("g14", raw, fps, nil, nil, nil)
comps, _, err := compositionsOf("g14", raw, fps, nil, nil)
if err != nil {
t.Fatal(err)
}
@@ -73,8 +73,7 @@ func TestAWakeWhileATakeInFollowsAddsNone(t *testing.T) {
return fmt.Errorf("%w: the key says building", errKeyWaits)
}
}
follower := make(chan struct{})
go func() { takeDoneKey(context.Background(), nil, "board"); close(follower) }()
go takeDoneKey(context.Background(), nil, "board")
time.Sleep(5 * time.Millisecond)
returned := make(chan struct{})
go func() { takeDoneKey(context.Background(), nil, "board"); close(returned) }()
@@ -84,101 +83,4 @@ func TestAWakeWhileATakeInFollowsAddsNone(t *testing.T) {
t.Fatal("a second wake waited for the follower")
}
close(release)
<-follower
}
// A wake while a take-in of the module follows it is not lost: the follower goes round once more after it succeeds, so
// a newer build's key written meanwhile is taken in (the review of issue 1945).
func TestAWakeDuringATakeInIsTakenAfterIt(t *testing.T) {
was, wasRetry, wasSay := doneKeyTakeOnce, doneKeyRetry, doneKeysSay
t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay = was, wasRetry, wasSay })
doneKeyRetry = 20 * time.Millisecond
doneKeysSay = func(string) {}
var mu sync.Mutex
calls := 0
inFirst := make(chan struct{})
release := make(chan struct{})
doneKeyTakeOnce = func(context.Context, *stores, string) error {
mu.Lock()
calls++
n := calls
mu.Unlock()
if n == 1 {
close(inFirst)
<-release
}
return nil
}
follower := make(chan struct{})
go func() { takeDoneKey(context.Background(), nil, "mesh-host"); close(follower) }()
<-inFirst
takeDoneKey(context.Background(), nil, "mesh-host") // the newer key's wake, while the first runs
close(release)
<-follower
mu.Lock()
defer mu.Unlock()
if calls != 2 {
t.Fatalf("taken %d times: the wake during the take-in was lost", calls)
}
}
// A wake that comes after the follower's last look but before it lets go of the module is not lost: the follower looks
// at the mark again once it has let go, and takes the module once more (the review of issue 1945).
func TestAWakeJustBeforeTheFollowerLetsGoIsTaken(t *testing.T) {
was, wasRetry, wasSay, wasHook := doneKeyTakeOnce, doneKeyRetry, doneKeysSay, doneKeysBeforeLetGo
t.Cleanup(func() {
doneKeyTakeOnce, doneKeyRetry, doneKeysSay, doneKeysBeforeLetGo = was, wasRetry, wasSay, wasHook
})
doneKeyRetry = 10 * time.Millisecond
doneKeysSay = func(string) {}
var mu sync.Mutex
calls := 0
doneKeyTakeOnce = func(context.Context, *stores, string) error {
mu.Lock()
calls++
mu.Unlock()
return nil
}
woke := false
doneKeysBeforeLetGo = func(module string) {
if !woke {
woke = true
takeDoneKey(context.Background(), nil, module) // fails its TryLock and marks the module
}
}
takeDoneKey(context.Background(), nil, "mesh-host")
mu.Lock()
defer mu.Unlock()
if calls != 2 {
t.Fatalf("taken %d times: the wake just before the follower let go was lost", calls)
}
}
// A wake during the start's take-in of a module is not lost: the start pass reads the mark once it lets go of the module,
// and hands the module to takeDoneKey, which takes it once more (the review of issue 1945: the third dropped wake of its
// kind).
func TestAWakeDuringTheStartsTakeInIsTaken(t *testing.T) {
was, wasRetry, wasSay := doneKeyTakeOnce, doneKeyRetry, doneKeysSay
t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay = was, wasRetry, wasSay })
doneKeyRetry = 10 * time.Millisecond
doneKeysSay = func(string) {}
var mu sync.Mutex
calls := 0
doneKeyTakeOnce = func(context.Context, *stores, string) error {
mu.Lock()
calls++
n := calls
mu.Unlock()
if n == 1 {
takeDoneKey(context.Background(), nil, "mesh-host") // the newer key's wake, during the start's take-in
}
return nil
}
takeDoneKeyAtStart(context.Background(), nil, "mesh-host")
background.Wait()
mu.Lock()
defer mu.Unlock()
if calls != 2 {
t.Fatalf("taken %d times: the wake during the start's take-in was lost", calls)
}
}
+54 -129
View File
@@ -26,52 +26,29 @@ import (
// every refusal is named in the error. It answers the keys it wrote, by node. Under the lease only: a controller that
// does not hold it writes nothing and says so.
func composeModules(ctx context.Context, open *stores, nodes []string) (map[string][]string, error) {
out, byNode := composeModulesByNode(ctx, open, nodes)
names := make([]string, 0, len(byNode))
for n := range byNode {
names = append(names, n)
}
sort.Strings(names)
var errs []error
for _, n := range names {
errs = append(errs, byNode[n])
}
return out, errors.Join(errs...)
}
// composeModulesByNode is composeModules with each node's error apart: a failure before any node is composed is every
// node's (the composer gives each caller only the errors of the nodes it asked for, issue 1945).
func composeModulesByNode(ctx context.Context, open *stores, nodes []string) (map[string][]string, map[string]error) {
all := func(err error) (map[string][]string, map[string]error) {
byNode := map[string]error{}
for _, n := range nodes {
byNode[n] = err
}
return nil, byNode
}
if _, err := theLease.epoch(ctx); err != nil {
return all(fmt.Errorf("nothing is composed: %w", err))
return nil, fmt.Errorf("nothing is composed: %w", err)
}
bus, err := aBus()
if err != nil {
return all(fmt.Errorf("nothing is composed: the bus: %w", err))
return nil, fmt.Errorf("nothing is composed: the bus: %w", err)
}
defer bus.Close()
js, err := jetstream.New(bus.Conn())
if err != nil {
return all(err)
return nil, err
}
sign, err := controllerSigner(ctx, open)
if err != nil {
return all(fmt.Errorf("nothing is composed: the controller's signing key: %w", err))
return nil, fmt.Errorf("nothing is composed: the controller's signing key: %w", err)
}
fps, err := currentFingerprints(ctx, open.inventory)
if err != nil {
return all(err)
return nil, err
}
entries, err := open.inventory.Catalogued(ctx)
if err != nil {
return all(err)
return nil, err
}
manifests := make(map[string]catalogue.Manifest, len(entries))
for _, e := range entries {
@@ -79,7 +56,7 @@ func composeModulesByNode(ctx context.Context, open *stores, nodes []string) (ma
}
upgrades, err := open.inventory.Upgrades(ctx)
if err != nil {
return all(err)
return nil, err
}
policy := func(module string) (string, string, string) {
u, ok := upgrades[module]
@@ -90,101 +67,59 @@ func composeModulesByNode(ctx context.Context, open *stores, nodes []string) (ma
}
sort.Strings(nodes)
out := map[string][]string{}
errsOf := map[string][]error{}
var errs []error
for _, node := range nodes {
// One composition of a node at a time in this process: two at once collided on its keys' revisions (issue 1945).
unlock := lockComposing(node)
func() {
defer unlock()
at, err := openComposedPlace(ctx, js, bus.Conn(), node)
if err != nil {
errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
return
}
at.sign, at.policy = sign, policy
if composeTrace != nil {
traceInput(ctx, open, node)
}
declared, err := composeSend(ctx, open, node)
if err != nil {
errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
return
}
body, err := compositionBody(declared)
if err != nil {
errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
return
}
// The provisioner's plans of the modules the node's composition hands to it, for the assignments mirror.
if err := open.inventory.RecordComposedPlans(ctx, node, plansKept(declared)); err != nil {
errsOf[node] = append(errsOf[node], fmt.Errorf("%s: its provisioner's plans: %w", node, err))
}
comps, moved, left, err := compositionsOf(node, body, fps, manifests, declared.modules, namedByOwnComposition(ctx, at))
if err != nil {
errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
return
}
if len(moved) > 0 {
fmt.Printf("compositions: %s's %s compose with the node's own: no fingerprint of the new form names them\n",
node, strings.Join(moved, ", "))
}
if len(left) > 0 {
fmt.Printf("compositions: %s's %s are left out of the node's own: the controller knows no fingerprint of the "+
"new form for them, and the node's want or have names their own composition (issue 1945)\n",
node, strings.Join(left, ", "))
}
if composeTrace != nil {
traceComposed(ctx, at, node, comps)
}
wrote, err := writeNodeCompositions(ctx, at, comps)
if len(wrote) > 0 {
out[node] = wrote
fmt.Printf("compositions: %s: wrote %s\n", node, strings.Join(wrote, ", "))
}
// And where a data-keeping provision bound each consumer: what the next resolution keeps it at (novox/hq ADR 0232).
if len(declared.Bindings) > 0 {
if err := open.inventory.RecordBindings(ctx, node, declared.Bindings); err != nil {
errsOf[node] = append(errsOf[node], fmt.Errorf("%s: where its consumers are bound to their data: %w", node, err))
}
}
if err != nil {
errsOf[node] = append(errsOf[node], fmt.Errorf("%s: %w", node, err))
}
}()
}
byNode := map[string]error{}
for node, errs := range errsOf {
if len(errs) > 0 {
byNode[node] = errors.Join(errs...)
at, err := openComposedPlace(ctx, js, bus.Conn(), node)
if err != nil {
errs = append(errs, fmt.Errorf("%s: %w", node, err))
continue
}
}
return out, byNode
}
// namedByOwnComposition says a module's want or have on the node names a composition of its own: a fingerprint of the
// new form (issue 1945).
func namedByOwnComposition(ctx context.Context, at composedPlace) func(module string) (bool, error) {
return func(module string) (bool, error) {
if at.wanted != nil {
fp, err := at.wanted(ctx, module)
if err != nil {
return false, err
}
if strings.HasPrefix(fp, sdkdelivery.SourcePrefix2) {
return true, nil
at.sign, at.policy = sign, policy
if composeTrace != nil {
traceInput(ctx, open, node)
}
declared, err := composeSend(ctx, open, node)
if err != nil {
errs = append(errs, fmt.Errorf("%s: %w", node, err))
continue
}
body, err := compositionBody(declared)
if err != nil {
errs = append(errs, fmt.Errorf("%s: %w", node, err))
continue
}
// The provisioner's plans of the modules the node's composition hands to it, for the assignments mirror.
if err := open.inventory.RecordComposedPlans(ctx, node, plansKept(declared)); err != nil {
errs = append(errs, fmt.Errorf("%s: its provisioner's plans: %w", node, err))
}
comps, moved, err := compositionsOf(node, body, fps, manifests, declared.modules)
if err != nil {
errs = append(errs, fmt.Errorf("%s: %w", node, err))
continue
}
if len(moved) > 0 {
fmt.Printf("compositions: %s's %s compose with the node's own: no fingerprint of the new form names them\n",
node, strings.Join(moved, ", "))
}
if composeTrace != nil {
traceComposed(ctx, at, node, comps)
}
wrote, err := writeNodeCompositions(ctx, at, comps)
if len(wrote) > 0 {
out[node] = wrote
fmt.Printf("compositions: %s: wrote %s\n", node, strings.Join(wrote, ", "))
}
// And where a data-keeping provision bound each consumer: what the next resolution keeps it at (novox/hq ADR 0232).
if len(declared.Bindings) > 0 {
if err := open.inventory.RecordBindings(ctx, node, declared.Bindings); err != nil {
errs = append(errs, fmt.Errorf("%s: where its consumers are bound to their data: %w", node, err))
}
}
if at.ran != nil {
h, found, err := at.ran(ctx, module)
if err != nil {
return false, err
}
if found && strings.HasPrefix(h.FP, sdkdelivery.SourcePrefix2) {
return true, nil
}
if err != nil {
errs = append(errs, fmt.Errorf("%s: %w", node, err))
}
return false, nil
}
return out, errors.Join(errs...)
}
// composeAndSay is composeModules for an act that has nothing to do with its answer but say it: what each former send
@@ -435,13 +370,3 @@ func compositionBody(declared sendable) ([]byte, error) {
declared.fileReferences = false
return declared.Body()
}
// composingLocks hold one composition of a node at a time in this process (issue 1945).
var composingLocks sync.Map
func lockComposing(node string) func() {
l, _ := composingLocks.LoadOrStore(node, &sync.Mutex{})
mu := l.(*sync.Mutex)
mu.Lock()
return mu.Unlock
}
+3 -3
View File
@@ -26,7 +26,7 @@ func TestALargeFileOfACompositionHasItsObjectInTheBucket(t *testing.T) {
if err != nil {
t.Fatal(err)
}
comps, _, _, err := compositionsOf("novox", body, nil, nil, nil, nil)
comps, _, err := compositionsOf("novox", body, nil, nil, nil)
if err != nil {
t.Fatal(err)
}
@@ -50,7 +50,7 @@ func TestACompositionNamingAnObjectTheBucketLacksIsRefused(t *testing.T) {
if err != nil {
t.Fatal(err)
}
comps, _, _, err := compositionsOf("novox", body, nil, nil, nil, nil)
comps, _, err := compositionsOf("novox", body, nil, nil, nil)
if err != nil {
t.Fatal(err)
}
@@ -78,7 +78,7 @@ func TestEveryFilledFileStaysFilledInTheCompositions(t *testing.T) {
for _, m := range []string{"power", "claude-code", "node-env"} {
delete(fps, m)
}
comps, _, _, err := compositionsOf(node, body, fps, nil, nil, nil)
comps, _, err := compositionsOf(node, body, fps, nil, nil)
if err != nil {
t.Fatalf("%s: %v", node, err)
}
+23 -72
View File
@@ -67,15 +67,8 @@ type composedPlace struct {
// fingerprint fps gives it, the node's own with its adoption and provisioner, and every resource of a module fps names
// no fingerprint for in the node's own, its module taken off it and named in moved. Data says the module's manifest
// declares data.
//
// **A module a want or a have names by its own composition is never in the node's own** (issue 1945): the node-engine
// runs that module's composition, and the same resources in the node's own were declared twice, which it refuses
// whole ("2 resources declare X", every node's apply failing from 2026-10-25 12:43). named says, for a module fps names
// no fingerprint of the new form for, that the node's want or have names a fingerprint of the new form; such a module's
// resources and filled paths are left out of this pass (its own composition, as written before, stays what the node
// runs) and it is named in left.
func compositionsOf(node string, body []byte, fps map[string]string, manifests map[string]catalogue.Manifest, runs []string,
named func(module string) (bool, error)) (out map[string]sdkdelivery.Composed, moved, left []string, err error) {
func compositionsOf(node string, body []byte, fps map[string]string, manifests map[string]catalogue.Manifest, runs []string) (
out map[string]sdkdelivery.Composed, moved []string, err error) {
var envelope struct {
Resources []map[string]json.RawMessage `json:"resources"`
@@ -84,9 +77,9 @@ func compositionsOf(node string, body []byte, fps map[string]string, manifests m
Provisioner string `json:"provisioner_account"`
}
if err := json.Unmarshal(body, &envelope); err != nil {
return nil, nil, nil, fmt.Errorf("%s's declaration: %w", node, err)
return nil, nil, fmt.Errorf("%s's declaration: %w", node, err)
}
inModule, ownComposition := map[string]bool{}, map[string]bool{}
inModule := map[string]bool{}
resources := make([]json.RawMessage, 0, len(envelope.Resources))
for _, r := range envelope.Resources {
var module string
@@ -94,30 +87,15 @@ func compositionsOf(node string, body []byte, fps map[string]string, manifests m
_ = json.Unmarshal(raw, &module)
}
if module != "" && !strings.HasPrefix(fps[module], sdkdelivery.SourcePrefix2) {
if !inModule[module] && !ownComposition[module] {
isNamed := false
if named != nil {
if isNamed, err = named(module); err != nil {
return nil, nil, nil, fmt.Errorf("whether %s's want or have names %s's own composition: %w",
node, module, err)
}
}
if isNamed {
ownComposition[module] = true
left = append(left, module)
} else {
inModule[module] = true
moved = append(moved, module)
}
}
if ownComposition[module] {
continue
}
delete(r, "module")
if !inModule[module] {
inModule[module] = true
moved = append(moved, module)
}
}
raw, err := json.Marshal(r)
if err != nil {
return nil, nil, nil, err
return nil, nil, err
}
resources = append(resources, raw)
}
@@ -128,26 +106,22 @@ func compositionsOf(node string, body []byte, fps map[string]string, manifests m
// and shanks).
var filled, ownFilled []string
for _, p := range envelope.Filled {
switch {
case filledByMoved(envelope.Resources, p, ownComposition):
// The file of a module whose own composition the node runs: that composition names it, not this pass.
case filledByMoved(envelope.Resources, p, inModule):
if filledByMoved(envelope.Resources, p, inModule) {
ownFilled = append(ownFilled, p)
default:
} else {
filled = append(filled, p)
}
}
sort.Strings(moved)
sort.Strings(left)
out, err = sdkdelivery.SplitComposed(node, resources, fps, filled,
sdkdelivery.NodeOwn{Adoption: envelope.Adoption, Provisioner: envelope.Provisioner})
if err != nil {
return nil, nil, nil, err
return nil, nil, err
}
if len(ownFilled) > 0 {
own, err := sdkdelivery.ComposedNodeKey(node)
if err != nil {
return nil, nil, nil, err
return nil, nil, err
}
c := out[own]
c.Filled = append(c.Filled, ownFilled...)
@@ -175,7 +149,7 @@ func compositionsOf(node string, body []byte, fps map[string]string, manifests m
out[key] = c
}
}
return out, moved, left, nil
return out, moved, nil
}
// filledByMoved says a filled path is a file of a module whose resources were moved into the node's own composition.
@@ -244,39 +218,16 @@ func writeComposed(ctx context.Context, at composedPlace, key string, c sdkdeliv
}
wrote = true
}
// **The digest kept is the digest of the composition the bucket holds** (issue 1945): another writer may write its
// composition and its digest between this one's composition and its digest, and the digest written last must not be
// of a composition the bucket no longer holds, or the assigner's want names it and the node applies nothing. After
// its digest, this writer reads the composition again and, when another holds the key, keeps that one's digest.
written := false
for follows := 0; ; follows++ {
judged, err := judgeMove(ctx, at, key, c, digest)
if composeTrace != nil {
composeTrace(fmt.Sprintf("%s: %s: written %v, judged %s %q user-list-only %v against %.19s, err %v", at.node, key,
wrote, judged.Judgement, judged.Words, judged.UserListOnly, judged.Against, err))
}
if err != nil {
return wrote || written, err
}
w, err := keepDigest(ctx, at.digests, key, judged)
written = written || w
if err != nil {
return wrote || written, err
}
now, rev, err := readComposed(ctx, at.bucket, key)
if err != nil {
return wrote || written, err
}
nowDigest, err := sdkdelivery.ComposedDigestOf(now)
if rev == 0 || err != nil || nowDigest == digest {
return wrote || written, nil
}
if follows >= 3 {
return wrote || written, fmt.Errorf("%s moved under this writer %d times; its digest is kept by the next pass",
key, follows+1)
}
c, digest = now, nowDigest
judged, err := judgeMove(ctx, at, key, c, digest)
if composeTrace != nil {
composeTrace(fmt.Sprintf("%s: %s: written %v, judged %s %q user-list-only %v against %.19s, err %v", at.node, key, wrote,
judged.Judgement, judged.Words, judged.UserListOnly, judged.Against, err))
}
if err != nil {
return wrote, err
}
written, err := keepDigest(ctx, at.digests, key, judged)
return wrote || written, err
}
// objectsHeld refuses a composition that names a file by a digest whose object the node's bucket does not hold: a node
+3 -54
View File
@@ -218,7 +218,7 @@ func TestTheCompositionsOfANodeAreItsDeclaration(t *testing.T) {
Resources []json.RawMessage `json:"resources"`
}
_ = json.Unmarshal(body, &d)
comps, moved, _, err := compositionsOf(node, body, fps, nil, nil, nil)
comps, moved, err := compositionsOf(node, body, fps, nil, nil)
if err != nil {
t.Fatalf("%s: %v", node, err)
}
@@ -253,7 +253,7 @@ func TestTheCompositionsOfANodeAreItsDeclaration(t *testing.T) {
// unchanged it writes nothing; a changed input writes a new revision of that key alone, by compare-and-set.
func TestACompositionIsWrittenOnceAndAgainOnlyWhenItChanges(t *testing.T) {
body, fps := capturedBody(t, "g14")
comps, _, _, err := compositionsOf("g14", body, fps, nil, nil, nil)
comps, _, err := compositionsOf("g14", body, fps, nil, nil)
if err != nil {
t.Fatal(err)
}
@@ -342,7 +342,7 @@ func TestCompositionsOutsideTheFiveNewestAreDeleted(t *testing.T) {
func TestAModuleWithNoResourceOfItsOwnStillHasItsComposition(t *testing.T) {
body := []byte(`{"declaration":1,"resources":[]}`)
fp := sdkdelivery.SourcePrefix2 + strings.Repeat("b", 64)
comps, _, _, err := compositionsOf("laptop", body, map[string]string{"board": fp}, nil, []string{"board", "legacy"}, nil)
comps, _, err := compositionsOf("laptop", body, map[string]string{"board": fp}, nil, []string{"board", "legacy"})
if err != nil {
t.Fatal(err)
}
@@ -360,54 +360,3 @@ func TestAModuleWithNoResourceOfItsOwnStillHasItsComposition(t *testing.T) {
}
}
}
// 1945/AC16 (unit): a module the controller knows no fingerprint of the new form for, whose want or have on the node
// names its own composition, is never in the node's own composition, neither its resources nor its filled paths, and is
// named as left; a module neither names still goes with the node's own, its filled path with it. (Live from 2026-10-25
// 12:43: zsh, power, restic and more were in both, and every node refused "2 resources declare X".)
func TestAModuleItsWantNamesIsNeverInTheNodesOwnComposition(t *testing.T) {
body := []byte(`{"declaration":1,"resources":[
{"id":"zsh.rc","type":"file","module":"zsh","path":"/etc/zsh/zshrc"},
{"id":"power.unit","type":"file","module":"power","path":"/etc/power.conf"},
{"id":"mesh-wireguard.conf","type":"file","module":"mesh-wireguard","path":"/etc/wireguard/mesh0.conf"}],
"filled_files":["/etc/zsh/zshrc","/etc/power.conf","/etc/wireguard/mesh0.conf"]}`)
asked := map[string]int{}
named := func(module string) (bool, error) {
asked[module]++
return module == "zsh" || module == "power", nil
}
comps, moved, left, err := compositionsOf("ace", body, map[string]string{}, nil, nil, named)
if err != nil {
t.Fatal(err)
}
if strings.Join(left, ",") != "power,zsh" || strings.Join(moved, ",") != "mesh-wireguard" {
t.Errorf("left %v, moved %v; want power,zsh left and mesh-wireguard moved", left, moved)
}
if asked["zsh"] != 1 || asked["power"] != 1 {
t.Errorf("asked each module once: %v", asked)
}
own, _ := sdkdelivery.ComposedNodeKey("ace")
for key, c := range comps {
for _, id := range resourceIDs(t, c.Resources) {
if strings.HasPrefix(id, "zsh.") || strings.HasPrefix(id, "power.") {
t.Errorf("%s holds %s, which the module's own composition declares", key, id)
}
}
for _, p := range c.Filled {
if p == "/etc/zsh/zshrc" || p == "/etc/power.conf" {
t.Errorf("%s names %s as filled, which the module's own composition names", key, p)
}
}
}
if ids := strings.Join(resourceIDs(t, comps[own].Resources), " "); ids != "mesh-wireguard.conf" {
t.Errorf("the node's own composition holds %q, want mesh-wireguard.conf alone", ids)
}
if f := strings.Join(comps[own].Filled, " "); f != "/etc/wireguard/mesh0.conf" {
t.Errorf("the node's own composition names %q as filled, want mesh-wireguard's file alone", f)
}
if _, _, _, err := compositionsOf("ace", body, map[string]string{}, nil, nil,
func(string) (bool, error) { return false, fmt.Errorf("the wants bucket is gone") }); err == nil {
t.Errorf("a want that cannot be read composed the node anyway")
}
}
-105
View File
@@ -1,105 +0,0 @@
package main
import (
"context"
"errors"
"sort"
"sync"
)
// **One composer, one pass at a time, each node once per pass** (issue 1945). A done key's take-in registers its module
// and asks the composer for its nodes; the composer gathers every node asked for while a pass runs and composes them in
// the next pass, in one composition of each node, then writes the grants once. Before, each take-in of a burst composed
// every node of its module itself, at once: they collided on the compositions' revisions ("wrong last sequence … key
// revision mismatch"), retried every 30 s, and the take-in that mattered waited behind them (2026-10-25 at 11:18 and
// 12:32). A caller waits for the pass that began after it asked, and hears that pass's error.
type composer struct {
composeNodes func(ctx context.Context, nodes []string) map[string]error
writeGrants func(ctx context.Context) error
mu sync.Mutex
pending map[string]bool
grants bool
waiters []waiter
wake chan struct{}
}
// waiter is a caller waiting for its pass: the nodes it asked for, whether it asked for the grants, and where its answer
// goes.
type waiter struct {
nodes []string
grants bool
done chan error
}
func newComposer(compose func(context.Context, []string) map[string]error, grants func(context.Context) error) *composer {
return &composer{composeNodes: compose, writeGrants: grants, pending: map[string]bool{}, wake: make(chan struct{}, 1)}
}
// compose asks for the nodes to be composed, and the grants written after when grants is true, and waits for the pass
// that does it, with the errors of its own nodes alone.
func (c *composer) compose(ctx context.Context, nodes []string, grants bool) error {
done := make(chan error, 1)
c.mu.Lock()
for _, n := range nodes {
c.pending[n] = true
}
c.grants = c.grants || grants
c.waiters = append(c.waiters, waiter{nodes: nodes, grants: grants, done: done})
c.mu.Unlock()
select {
case c.wake <- struct{}{}:
default:
}
select {
case err := <-done:
return err
case <-ctx.Done():
return ctx.Err()
}
}
// run makes the passes until ctx ends: each takes everything asked for since the last began.
func (c *composer) run(ctx context.Context) {
for {
select {
case <-ctx.Done():
return
case <-c.wake:
}
c.mu.Lock()
nodes := make([]string, 0, len(c.pending))
for n := range c.pending {
nodes = append(nodes, n)
}
grants, waiters := c.grants, c.waiters
c.pending, c.grants, c.waiters = map[string]bool{}, false, nil
c.mu.Unlock()
if len(waiters) == 0 {
continue
}
sort.Strings(nodes)
var failed map[string]error
if len(nodes) > 0 {
failed = c.composeNodes(ctx, nodes)
}
// The grants whenever a caller asked for them, whatever a node's composition said: a node that failed fails its
// own callers alone (issue 1945).
var grantsErr error
if grants {
grantsErr = c.writeGrants(ctx)
}
for _, w := range waiters {
var errs []error
for _, n := range w.nodes {
if err := failed[n]; err != nil {
errs = append(errs, err)
}
}
if w.grants && grantsErr != nil {
errs = append(errs, grantsErr)
}
w.done <- errors.Join(errs...)
}
}
}
-143
View File
@@ -1,143 +0,0 @@
package main
import (
"context"
"slices"
"sort"
"sync"
"testing"
"time"
)
// A burst of done keys composes each node once per pass and writes the grants once, never two passes at a time: the
// take-ins of a burst each composed every node of their module at once, collided on the compositions' revisions and
// retried every 30 s, while the one that mattered (mesh-host's) waited (issue 1945, live on 2026-10-25 at 11:18 and
// 12:32).
func TestABurstComposesEachNodeOncePerPass(t *testing.T) {
var mu sync.Mutex
var passes [][]string
grants := 0
running, maxRunning := 0, 0
release := make(chan struct{})
first := true
c := newComposer(func(_ context.Context, nodes []string) map[string]error {
mu.Lock()
running++
maxRunning = max(maxRunning, running)
passes = append(passes, slices.Clone(nodes))
wait := first
first = false
mu.Unlock()
if wait {
<-release // the first pass runs while the rest of the burst arrives
}
mu.Lock()
running--
mu.Unlock()
return nil
}, func(context.Context) error {
mu.Lock()
grants++
mu.Unlock()
return nil
})
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go c.run(ctx)
var wg sync.WaitGroup
wg.Add(1)
go func() { defer wg.Done(); _ = c.compose(ctx, []string{"ace"}, true) }()
time.Sleep(20 * time.Millisecond)
for _, nodes := range [][]string{{"ace", "g14"}, {"g14"}, {"shanks", "ace"}, {"novox"}, {"ace"}} {
wg.Add(1)
go func() { defer wg.Done(); _ = c.compose(ctx, nodes, true) }()
}
time.Sleep(20 * time.Millisecond)
close(release)
wg.Wait()
mu.Lock()
defer mu.Unlock()
if maxRunning != 1 {
t.Fatalf("%d passes ran at once", maxRunning)
}
if len(passes) != 2 {
t.Fatalf("passes %v: want the first and one more for the rest of the burst", passes)
}
second := slices.Clone(passes[1])
sort.Strings(second)
if !slices.Equal(second, []string{"ace", "g14", "novox", "shanks"}) {
t.Fatalf("the second pass composed %v: each node once", passes[1])
}
if grants != 2 {
t.Fatalf("the grants were written %d times for two passes", grants)
}
}
// A caller hears its own pass's failure, and only a pass that began after it asked counts as its own.
func TestACallerHearsThePassThatComposedItsNodes(t *testing.T) {
calls := 0
c := newComposer(func(_ context.Context, nodes []string) map[string]error {
calls++
if calls == 1 {
return map[string]error{"ace": context.DeadlineExceeded}
}
return nil
}, func(context.Context) error { return nil })
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go c.run(ctx)
if err := c.compose(ctx, []string{"ace"}, false); err == nil {
t.Fatal("a failed pass was not heard")
}
if err := c.compose(ctx, []string{"ace"}, false); err != nil {
t.Fatalf("a good pass said %v", err)
}
}
// One node that fails its composition fails only the callers that asked for it, and the grants are written whenever a
// caller asked for them (the review of issue 1945: one joined error failed every take-in of the pass and stopped their
// grants).
func TestANodeThatFailsFailsOnlyItsCallers(t *testing.T) {
grants := 0
release := make(chan struct{})
first := true
c := newComposer(func(_ context.Context, nodes []string) map[string]error {
if first {
first = false
<-release
}
out := map[string]error{}
for _, n := range nodes {
if n == "broken" {
out[n] = context.DeadlineExceeded
}
}
return out
}, func(context.Context) error { grants++; return nil })
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go c.run(ctx)
go func() { _ = c.compose(ctx, []string{"warm"}, false) }()
time.Sleep(20 * time.Millisecond)
errs := make(chan error, 2)
go func() { errs <- c.compose(ctx, []string{"ace"}, true) }()
go func() { errs <- c.compose(ctx, []string{"broken"}, false) }()
time.Sleep(20 * time.Millisecond)
close(release)
var failed, fine int
for range 2 {
if err := <-errs; err != nil {
failed++
} else {
fine++
}
}
if failed != 1 || fine != 1 {
t.Fatalf("failed %d, fine %d: only the caller of the broken node fails", failed, fine)
}
if grants != 1 {
t.Fatalf("the grants were written %d times; a caller asked for them", grants)
}
}
@@ -1,93 +0,0 @@
package main
import (
"context"
"errors"
"fmt"
"testing"
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/novox/mesh-controller/internal/builderkeys"
)
// 1945/AC15 (unit): a module registered at a commit no build record points at (a catalogue merge re-indexed it with
// the fingerprint it had, and the take-in registered it from the done key whose record stands at its first commit)
// has, as its current fingerprint, the one the indexer's trunk value at that commit names; a module whose commit has a
// build keeps that build's; a module the indexer names nothing of the new form for has none; an indexer error is the
// answer's, never a module silently left out.
func TestACommitWithNoBuildOfItsOwnTakesTheIndexersFingerprint(t *testing.T) {
zsh, board := grantFP('a'), grantFP('b')
commits := map[string]string{"zsh": "68f7431d98d7", "board": "2167b95a1111", "legacy": "abcdef012345", "held": ""}
built := func(_ context.Context, module, commit string) (string, bool, error) {
if module == "board" {
return board, true, nil
}
return "", false, nil
}
indexed := func(_ context.Context, module, commit string) (string, error) {
switch module {
case "zsh":
return zsh, nil
case "board":
t.Errorf("the indexer was asked for board, whose commit has a build")
case "held":
t.Errorf("the indexer was asked for a module with no commit")
}
return "", nil
}
got, err := fingerprintsAt(context.Background(), commits, built, indexed)
if err != nil {
t.Fatal(err)
}
if got["zsh"] != zsh || got["board"] != board || len(got) != 2 {
t.Errorf("got %v, want zsh at %s and board at %s alone", got, zsh, board)
}
broken := func(context.Context, string, string) (string, error) { return "", errors.New("the bus is gone") }
if _, err := fingerprintsAt(context.Background(), commits, built, broken); err == nil {
t.Errorf("an indexer error was not the answer's")
}
}
// fakeTrunk is the indexer's trunk values of one module: now, and its history by commit.
type fakeTrunk struct {
now *sdkdelivery.Module
history map[string]sdkdelivery.Module
asked []string
}
func (f *fakeTrunk) ModuleAt(_ context.Context, module, commit string) (sdkdelivery.Module, error) {
f.asked = append(f.asked, commit)
if commit == "" {
if f.now == nil {
return sdkdelivery.Module{}, fmt.Errorf("%w (%s)", builderkeys.ErrNoValue, module)
}
return *f.now, nil
}
m, ok := f.history[commit]
if !ok {
return sdkdelivery.Module{}, fmt.Errorf("%w (%s at %s)", builderkeys.ErrNoValue, module, commit)
}
return m, nil
}
// 1945/AC15 (unit): the indexer's fingerprint at a commit is read from the value it holds now when that is the
// commit's (one read), else from the key's history at the commit; none held is no fingerprint.
func TestTheIndexersFingerprintAtACommit(t *testing.T) {
now := sdkdelivery.Module{Commit: "68f7431d98d7a1279cba3a615b6e9f22c3295783", FP2: grantFP('a')}
f := &fakeTrunk{now: &now}
if fp, err := indexedFingerprintAt(context.Background(), f, "zsh", "68f7431d98d7a1279cba3a615b6e9f22c3295783"); err != nil || fp != grantFP('a') || len(f.asked) != 1 {
t.Errorf("now's commit: %q (%v), asked %v", fp, err, f.asked)
}
f = &fakeTrunk{now: &now, history: map[string]sdkdelivery.Module{"2167b95a": {Commit: "2167b95a", FP2: grantFP('c')}}}
if fp, err := indexedFingerprintAt(context.Background(), f, "zsh", "2167b95a"); err != nil || fp != grantFP('c') {
t.Errorf("an older commit: %q (%v)", fp, err)
}
if fp, err := indexedFingerprintAt(context.Background(), f, "zsh", "99999999"); err != nil || fp != "" {
t.Errorf("a commit the indexer never wrote: %q (%v)", fp, err)
}
if fp, err := indexedFingerprintAt(context.Background(), &fakeTrunk{}, "zsh", "68f7431d"); err != nil || fp != "" {
t.Errorf("no value at all: %q (%v)", fp, err)
}
}
+2 -51
View File
@@ -53,61 +53,12 @@ func startedOn(h sdkdelivery.Have, written time.Time) time.Time {
const givenWhy = "a value given by hand lives only until its module's first good start under the mesh (novox/hq ADR 0228)"
// replaceGiven replaces what a module's good start makes due, composes the machines, and says so. One at a time for a
// node: a call that finds one running leaves what it was asked for as a mark, and the one running takes it once it is
// done, so a start heard meanwhile is never lost (issue 1945, the kind of the dropped wake).
// replaceGiven replaces what a module's good start makes due, composes the machines, and says so.
func replaceGiven(ctx context.Context, open *stores, node string, started map[string]time.Time) {
if _, busy := replacing.LoadOrStore(node, true); busy {
markGivenAgain(node, started)
return
}
for {
replaceGivenOnce(ctx, open, node, started)
replacing.Delete(node)
// Read after letting go: a call meanwhile only marked the node.
pending := takeGivenMark(node)
if pending == nil || ctx.Err() != nil {
return
}
if _, busy := replacing.LoadOrStore(node, true); busy {
markGivenAgain(node, pending) // the one running now takes it
return
}
started = pending
}
}
// givenAgain is, by node, the starts asked for while a replacement ran there, merged.
var givenAgain = struct {
sync.Mutex
by map[string]map[string]time.Time
}{by: map[string]map[string]time.Time{}}
func markGivenAgain(node string, started map[string]time.Time) {
givenAgain.Lock()
defer givenAgain.Unlock()
m := givenAgain.by[node]
if m == nil {
m = map[string]time.Time{}
givenAgain.by[node] = m
}
for module, at := range started {
if at.After(m[module]) {
m[module] = at
}
}
}
func takeGivenMark(node string) map[string]time.Time {
givenAgain.Lock()
defer givenAgain.Unlock()
m := givenAgain.by[node]
delete(givenAgain.by, node)
return m
}
// replaceGivenOnce is one replacement on a node; a variable so a test can stand in for it.
var replaceGivenOnce = func(ctx context.Context, open *stores, node string, started map[string]time.Time) {
defer replacing.Delete(node)
replaced, err := open.inventory.ReplaceGivenAfterStart(ctx, node, started)
if err != nil {
log.Printf("a value given by hand on %s could not be replaced after its module started: %v", node, err)
-34
View File
@@ -1,34 +0,0 @@
package main
import (
"context"
"sync"
"testing"
"time"
)
// A start heard while a replacement of given values runs on the node is not lost: it is marked, and the replacement
// running takes it once it is done (issue 1945, the kind of the dropped wake: replaceGiven returned at once when busy).
func TestAStartHeardDuringAReplacementIsTaken(t *testing.T) {
was := replaceGivenOnce
t.Cleanup(func() { replaceGivenOnce = was })
var mu sync.Mutex
var asked []map[string]time.Time
at := time.Date(2026, 10, 25, 13, 0, 0, 0, time.UTC)
replaceGivenOnce = func(ctx context.Context, open *stores, node string, started map[string]time.Time) {
mu.Lock()
asked = append(asked, started)
first := len(asked) == 1
mu.Unlock()
if first {
// Another module starts on the node while this one is being replaced.
replaceGiven(ctx, open, node, map[string]time.Time{"redis": at})
}
}
replaceGiven(context.Background(), nil, "ace", map[string]time.Time{"postgres": at})
mu.Lock()
defer mu.Unlock()
if len(asked) != 2 || !asked[1]["redis"].Equal(at) {
t.Fatalf("asked %v: the start heard during the replacement was lost", asked)
}
}
+8 -108
View File
@@ -11,7 +11,6 @@ import (
"github.com/nats-io/nats.go/jetstream"
"github.com/novox/mesh-controller/internal/broker"
"github.com/novox/mesh-controller/internal/builderkeys"
"github.com/novox/mesh-controller/internal/inventory"
)
@@ -56,128 +55,29 @@ func grantsFrom(principals []broker.Principal, hashes map[string]string, current
// currentFingerprints is, by module, the fingerprint of the build each registered module is registered at: its
// manifest's commit (built_from), that commit's newest successful build, and the fingerprint that build names.
//
// **A commit with no build of its own asks the indexer** (issue 1945, live from 2026-10-25 12:43). A catalogue merge
// that leaves a module's sources alone re-indexes it at the merge's commit with the fingerprint it had: the take-in
// registers the module at the new commit from the same done key, whose build record (one per key and revision) already
// stands at the commit it was first taken in at, so the new commit has no build pointing at it. Read only by the
// commit, such a module had no fingerprint of the new form: it was composed into the node's own composition while its
// want named its own, and every node declared its files twice. The indexer's trunk value at the commit names the
// fingerprint the module was registered from, so that is its fingerprint.
func currentFingerprints(ctx context.Context, inv *inventory.Inventory) (map[string]string, error) {
builds, err := inv.CurrentBuilds(ctx)
if err != nil {
return nil, err
}
commits := make(map[string]string, len(builds))
for module, b := range builds {
commits[module] = b.Commit
}
var indexed func(ctx context.Context, module, commit string) (string, error)
var closeIndexed func()
defer func() {
if closeIndexed != nil {
closeIndexed()
}
}()
return fingerprintsAt(ctx, commits,
func(ctx context.Context, module, commit string) (string, bool, error) {
id, err := inv.BuildOf(ctx, module, commit)
if err != nil || id == "" {
return "", false, nil
}
build, found, err := inv.BuildByID(ctx, id)
if err != nil || !found {
return "", false, nil
}
return build.SourceFingerprint, true, nil
},
func(ctx context.Context, module, commit string) (string, error) {
if indexed == nil {
read, done, err := openIndexedFingerprints()
if err != nil {
return "", err
}
indexed, closeIndexed = read, done
}
return indexed(ctx, module, commit)
})
}
// fingerprintsAt is, by module, the fingerprint of the build registered at its commit (built), or else the fingerprint
// of the new form the indexer's trunk value at that commit names (indexed). A module with no commit (held without a
// source) has none, and a module neither knows has none. An error of the indexer's is the answer's: a module left out
// for it would be composed into the node's own composition beside its own (issue 1945).
func fingerprintsAt(ctx context.Context, commits map[string]string,
built func(ctx context.Context, module, commit string) (string, bool, error),
indexed func(ctx context.Context, module, commit string) (string, error)) (map[string]string, error) {
modules := make([]string, 0, len(commits))
for module := range commits {
modules = append(modules, module)
}
sort.Strings(modules)
out := map[string]string{}
for _, module := range modules {
commit := commits[module]
if commit == "" {
for module, b := range builds {
if b.Commit == "" {
continue
}
fp, found, err := built(ctx, module, commit)
if err != nil {
return nil, err
}
if found {
out[module] = fp
id, err := inv.BuildOf(ctx, module, b.Commit)
if err != nil || id == "" {
continue
}
fp, err = indexed(ctx, module, commit)
if err != nil {
return nil, fmt.Errorf("the fingerprint %s is registered at (%.8s): %w", module, commit, err)
}
if strings.HasPrefix(fp, sdkdelivery.SourcePrefix2) {
out[module] = fp
build, found, err := inv.BuildByID(ctx, id)
if err != nil || !found {
continue
}
out[module] = build.SourceFingerprint
}
return out, nil
}
// openIndexedFingerprints opens the indexer's trunk values: what reads the fingerprint of the new form a module's value
// at a commit names ("" when it holds none for that commit), and what closes the bus connection it opened.
var openIndexedFingerprints = func() (func(ctx context.Context, module, commit string) (string, error), func(), error) {
keys, closeKeys, err := openBuilderKeys()
if err != nil {
return nil, nil, fmt.Errorf("the indexer's trunk values: %w", err)
}
return func(ctx context.Context, module, commit string) (string, error) {
return indexedFingerprintAt(ctx, keys, module, commit)
}, closeKeys, nil
}
// indexedFingerprintAt is the fingerprint of the new form the indexer's value of module at commit names: the value it
// holds now when that is the commit's (one read), else the newest of the key's history at the commit; "" when none.
func indexedFingerprintAt(ctx context.Context, keys interface {
ModuleAt(ctx context.Context, module, commit string) (sdkdelivery.Module, error)
}, module, commit string) (string, error) {
m, err := keys.ModuleAt(ctx, module, "")
if err == nil && sameCommitID(m.Commit, commit) {
return m.FP2, nil
}
if err != nil && !errors.Is(err, builderkeys.ErrNoValue) {
return "", err
}
if errors.Is(err, builderkeys.ErrNoValue) {
return "", nil
}
m, err = keys.ModuleAt(ctx, module, commit)
if errors.Is(err, builderkeys.ErrNoValue) {
return "", nil
}
if err != nil {
return "", err
}
return m.FP2, nil
}
// grantsKV is what writing the grants needs of their bucket.
type grantsKV interface {
Get(ctx context.Context, key string) (jetstream.KeyValueEntry, error)