Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 jschoubben
|
98284f15ca |
@@ -399,8 +399,6 @@ func blockedElsewhere(ctx context.Context, open *stores, except string) string {
|
||||
// purpose, never by re-assigning); and when the bus cannot be reached from here, the line names the
|
||||
// verb and the push that would refuse the module until it is run — never a silent placeholder (novox/hq issue 203).
|
||||
func issueOnAssign(ctx context.Context, open *stores, node, module string) string {
|
||||
// Mint only what no row names: the assignments' watch hears the same write and may issue at once (ADR 0543).
|
||||
ctx = onlyIfAbsent(ctx)
|
||||
inv := open.inventory
|
||||
shelf, err := inv.Catalogue(ctx)
|
||||
if err != nil {
|
||||
|
||||
@@ -12,7 +12,6 @@ import (
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
@@ -28,7 +27,7 @@ import (
|
||||
)
|
||||
|
||||
// askerFrom is the serving controller's asker; nil in any other process.
|
||||
var askerFrom atomic.Pointer[asker]
|
||||
var askerFrom *asker
|
||||
|
||||
// askWithin is how long a verb a warrant chose is given to answer.
|
||||
const askWithin = time.Minute
|
||||
@@ -574,8 +573,8 @@ func startAsking(ctx context.Context, open *stores, server *link.Server, conn *n
|
||||
fmt.Printf("the operator's answers cannot be heard, so nothing is asked: %v\n", err)
|
||||
return
|
||||
}
|
||||
askerFrom.Store(a)
|
||||
goServing(ctx, func() { a.keep(ctx) })
|
||||
askerFrom = a
|
||||
go a.keep(ctx)
|
||||
}
|
||||
|
||||
// servingAsker is the serving controller's asker, wired to the mesh: what it reads, publishes, performs and
|
||||
|
||||
@@ -1,387 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
"github.com/novox/mesh-controller/internal/chain"
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
)
|
||||
|
||||
// An assignment that stands issues its node's bus accounts and the module's grant (novox/hq ADR 0543, issue 1945; a
|
||||
// step of research 072's grants: the memberships and credentials heard from placement, not from a send or a build).
|
||||
//
|
||||
// ADR 0539 clause 7 issues a module's credential and grant when its build is done. A module already built and assigned
|
||||
// to one more node never has a build done again, so nothing issued that node's account: on 2026-10-25 at 17:19
|
||||
// mesh-build-node and mesh-build-go were assigned to ace, composed for it, and the bus's user list never named
|
||||
// ace.mesh-build-node nor ace.mesh-build-go; one runtime found no credential, the other was refused by the bus, both
|
||||
// restarting every 5 s. The assign act composes the assigned node alone (composedAfterAct): never the node holding the
|
||||
// bus, never the grant.
|
||||
//
|
||||
// So the write of an assignments key is the event. When a key `<node>.<module>` of the assignments bucket comes to stand
|
||||
// (a new key, or one that was ended or pending), the lease holder issues the module's accounts on that node that are not
|
||||
// minted (idempotent: a minted account is never issued again, a rotation is on purpose), composes the node holding the
|
||||
// bus so its user list carries them, and writes the grants. A rewrite of an assignment that already stood (the
|
||||
// provisioner's plan kept in its value) does nothing. A starting controller looks at every standing assignment once,
|
||||
// from one reading of the catalogue: it issues what is missing, then composes the bus's node and writes the grants
|
||||
// once. The assign act issues too; both mint only an account no row names, in one statement, so exactly one of them
|
||||
// mints and seals it (mintBusUser, the review of issue 1945).
|
||||
|
||||
// assignmentAccountSteps are what an assignment that comes to stand does, each a step a test may stand in for.
|
||||
type assignmentAccountSteps struct {
|
||||
// holds says this controller holds the lease: only its holder issues.
|
||||
holds func(ctx context.Context) bool
|
||||
// catalogue reads the catalogue once, for every assignment a look issues for.
|
||||
catalogue func(ctx context.Context) (map[string]catalogue.Manifest, error)
|
||||
// issue issues the module's accounts on the node that are not minted, from the catalogue read; issued says it
|
||||
// minted one, line what it did.
|
||||
issue func(ctx context.Context, shelf map[string]catalogue.Manifest, node, module string) (issued bool, line string, err error)
|
||||
// busHolder is the node holding the bus, "" for none.
|
||||
busHolder func(ctx context.Context) (string, error)
|
||||
// compose writes the nodes' compositions; grants writes the grants issued.
|
||||
compose func(ctx context.Context, nodes []string) error
|
||||
grants func(ctx context.Context) error
|
||||
}
|
||||
|
||||
// settle issues what the standing assignment of module on node lacks. fresh says the assignment just came to stand: the
|
||||
// bus's node is composed and the grants written even when every account was minted before (a module assigned again
|
||||
// keeps its account, and the bus's user list left it out while it was unassigned). It answers what it did, "" for
|
||||
// nothing.
|
||||
func (s assignmentAccountSteps) settle(ctx context.Context, node, module string) (string, error) {
|
||||
if module == controllerModule {
|
||||
// The controller's own module is issued at the terminal alone: its `broker` secret is the credential it is
|
||||
// using (rollout.go), and it is assigned and unassigned there only (refusedMovingTheController).
|
||||
return "", nil
|
||||
}
|
||||
if !s.holds(ctx) {
|
||||
return "", errNotHolding
|
||||
}
|
||||
shelf, err := s.catalogue(ctx)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("the catalogue cannot be read: %w", err)
|
||||
}
|
||||
_, line, err := s.issue(ctx, shelf, node, module)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("its bus accounts on %s are not issued: %w", node, err)
|
||||
}
|
||||
said, err := s.cover(ctx)
|
||||
prefix := module + " on " + node + ": "
|
||||
if line != "" {
|
||||
prefix += line + "; "
|
||||
}
|
||||
return prefix + said, err
|
||||
}
|
||||
|
||||
// look is the start's look: the accounts every standing assignment lacks, issued one after another from one reading of
|
||||
// the catalogue, then the bus's node composed and the grants written once, when it looked at any assignment, whatever
|
||||
// it issued (an assignment that came to stand while no controller heard it may have its accounts and still lack its
|
||||
// place in the bus's user list or its grant). It answers what it did.
|
||||
func (s assignmentAccountSteps) look(ctx context.Context, standing []chain.Assignment) (string, error) {
|
||||
var looked []chain.Assignment
|
||||
for _, a := range standing {
|
||||
if a.Module != controllerModule {
|
||||
looked = append(looked, a)
|
||||
}
|
||||
}
|
||||
if len(looked) == 0 {
|
||||
return "", nil
|
||||
}
|
||||
if !s.holds(ctx) {
|
||||
return "", errNotHolding
|
||||
}
|
||||
shelf, err := s.catalogue(ctx)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("the catalogue cannot be read: %w", err)
|
||||
}
|
||||
var lines []string
|
||||
for _, a := range looked {
|
||||
if ctx.Err() != nil {
|
||||
return "", ctx.Err()
|
||||
}
|
||||
if _, line, err := s.issue(ctx, shelf, a.Node, a.Module); err != nil {
|
||||
return "", fmt.Errorf("%s's bus accounts on %s are not issued: %w", a.Module, a.Node, err)
|
||||
} else if line != "" {
|
||||
lines = append(lines, a.Module+" on "+a.Node+": "+line)
|
||||
}
|
||||
}
|
||||
said, err := s.cover(ctx)
|
||||
lines = append(lines, fmt.Sprintf("%d standing assignment(s) looked at; %s", len(looked), said))
|
||||
return strings.Join(lines, "\n"), err
|
||||
}
|
||||
|
||||
// errNotHolding is a controller that does not hold the lease: it issues nothing, and tries again until it holds it.
|
||||
var errNotHolding = errors.New("this controller does not hold the lease")
|
||||
|
||||
// cover composes the node holding the bus and writes the grants, and says so.
|
||||
func (s assignmentAccountSteps) cover(ctx context.Context) (string, error) {
|
||||
holder, err := s.busHolder(ctx)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("the node holding the bus cannot be read: %w", err)
|
||||
}
|
||||
if holder != "" {
|
||||
if err := s.compose(ctx, []string{holder}); err != nil {
|
||||
return "", fmt.Errorf("the bus's user list on %s is not composed: %w", holder, err)
|
||||
}
|
||||
}
|
||||
if err := s.grants(ctx); err != nil {
|
||||
return "", err
|
||||
}
|
||||
if holder != "" {
|
||||
return "the bus's user list composed on " + holder + ", the grants written", nil
|
||||
}
|
||||
return "the grants written", nil
|
||||
}
|
||||
|
||||
// assignmentStands says a value of the assignments bucket holds an assignment that stands: made (a module only pulled in
|
||||
// by another names no time assigned), not ended and in no state (chain.Assignment.Stands).
|
||||
func assignmentStands(a chain.Assignment) bool {
|
||||
return !a.Assigned.IsZero() && a.Stands() && a.Node != "" && a.Module != ""
|
||||
}
|
||||
|
||||
// assignmentTurns keeps which keys of the assignments bucket stood when last heard, to tell an assignment that comes to
|
||||
// stand from a rewrite of one that stood.
|
||||
type assignmentTurns struct {
|
||||
mu sync.Mutex
|
||||
stood map[string]bool
|
||||
}
|
||||
|
||||
// heard records a value of key (nil for a delete or a purge) and answers whether it holds an assignment that just came
|
||||
// to stand.
|
||||
func (t *assignmentTurns) heard(key string, value []byte) (chain.Assignment, bool) {
|
||||
var a chain.Assignment
|
||||
stands := false
|
||||
if value != nil && json.Unmarshal(value, &a) == nil {
|
||||
stands = assignmentStands(a)
|
||||
}
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
if t.stood == nil {
|
||||
t.stood = map[string]bool{}
|
||||
}
|
||||
was := t.stood[key]
|
||||
t.stood[key] = stands
|
||||
return a, stands && !was
|
||||
}
|
||||
|
||||
// ownAccountOf is the username of the account a module speaks on the bus with on node, its own and no process's: the
|
||||
// module's when it reads a `broker` own secret, else its provisioner client's when it runs one; false when it has none
|
||||
// (a module whose accounts are its processes' alone, or one that does not speak on the bus).
|
||||
func ownAccountOf(m catalogue.Manifest, node string) (string, bool) {
|
||||
if _, reads := m.OwnSecrets["broker"]; reads {
|
||||
return broker.Principal{Kind: broker.KindModule, Node: node, Module: m.Module}.Username(), true
|
||||
}
|
||||
if runsTheProvisionerClient(m) {
|
||||
return broker.Principal{Kind: broker.KindProvisioner, Node: node, Module: m.Module}.Username(), true
|
||||
}
|
||||
return "", false
|
||||
}
|
||||
|
||||
// issueAssignmentAccounts issues module's accounts on node that are not minted: its own (issueOnTheNewBus) and its
|
||||
// processes' (issueMissingProcessAccounts). A module the catalogue does not hold, or one with nowhere to deliver an
|
||||
// account (mayIssue), has nothing to issue; an account minted before is kept.
|
||||
func issueAssignmentAccounts(ctx context.Context, inv *inventory.Inventory, shelf map[string]catalogue.Manifest,
|
||||
node, module string) (bool, string, error) {
|
||||
// Mint only what no row names: the assign act hears its own write and may issue at once (the review of issue 1945).
|
||||
ctx = onlyIfAbsent(ctx)
|
||||
m, known := shelf[module]
|
||||
if !known || mayIssue(m) != nil {
|
||||
return false, "", nil
|
||||
}
|
||||
if user, own := ownAccountOf(m, node); own {
|
||||
_, minted, err := inv.BusUserHash(ctx, user)
|
||||
if err != nil {
|
||||
return false, "", fmt.Errorf("whether %s is minted cannot be read: %w", user, err)
|
||||
}
|
||||
if !minted {
|
||||
busAddress, err := broker.BusAddress()
|
||||
if err != nil {
|
||||
return false, "", err
|
||||
}
|
||||
// Its processes' accounts too: issueOnTheNewBus issues every one of them with the module's own.
|
||||
if err := issueOnTheNewBus(ctx, inv, m, node, busAddress); err != nil {
|
||||
return false, "", err
|
||||
}
|
||||
return true, "its bus credential issued as " + user, nil
|
||||
}
|
||||
}
|
||||
line, err := issueMissingProcessAccounts(ctx, inv, m, node)
|
||||
if err != nil {
|
||||
return false, "", err
|
||||
}
|
||||
return line != "", line, nil
|
||||
}
|
||||
|
||||
// liveAssignmentAccountSteps are the steps on the controller's own stores and the bus: the composer's passes, as the
|
||||
// done keys' take-ins use them, so a node asked for by both is composed once a pass.
|
||||
func liveAssignmentAccountSteps(open *stores) assignmentAccountSteps {
|
||||
inv := open.inventory
|
||||
return assignmentAccountSteps{
|
||||
holds: func(ctx context.Context) bool {
|
||||
// The lease's gate, as every act's: a serving controller the bus does not grant the lease yet acts unleased.
|
||||
_, err := theLease.epoch(ctx)
|
||||
return err == nil
|
||||
},
|
||||
catalogue: func(ctx context.Context) (map[string]catalogue.Manifest, error) { return inv.Catalogue(ctx) },
|
||||
issue: func(ctx context.Context, shelf map[string]catalogue.Manifest, node, module string) (bool, string, error) {
|
||||
return issueAssignmentAccounts(ctx, inv, shelf, node, module)
|
||||
},
|
||||
busHolder: func(ctx context.Context) (string, error) {
|
||||
holders, err := seatHolders(ctx, inv)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return holders[catalogue.BrokerSeat].Node, nil
|
||||
},
|
||||
compose: func(ctx context.Context, nodes []string) error {
|
||||
return composerFor(open).compose(ctx, nodes, false)
|
||||
},
|
||||
grants: func(ctx context.Context) error {
|
||||
return composerFor(open).compose(ctx, nil, true)
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// assignmentAccountsRetry is how long a settle that failed waits before it tries again.
|
||||
var assignmentAccountsRetry = 30 * time.Second
|
||||
|
||||
// untilSettled runs settle until it works or ctx ends: a failure, or a controller not holding the lease, is said and
|
||||
// tried again every assignmentAccountsRetry, never dropped, since no later write of the key may come.
|
||||
func untilSettled(ctx context.Context, what string, settle func(context.Context) (string, error)) {
|
||||
for {
|
||||
said, err := settle(ctx)
|
||||
if err == nil {
|
||||
if said != "" {
|
||||
for _, line := range strings.Split(said, "\n") {
|
||||
fmt.Printf("assignments: %s\n", line)
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
fmt.Printf("assignments: %s: %v; tried again in %s\n", what, err, assignmentAccountsRetry)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-time.After(assignmentAccountsRetry):
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// watchAssignmentAccounts settles every assignment of the assignments bucket that comes to stand, for as long as ctx: the
|
||||
// bucket's current values first (the start's look, settled as not fresh), then each write. A watch that ends, or a
|
||||
// bucket not there yet, is watched again a minute later.
|
||||
func watchAssignmentAccounts(ctx context.Context, open *stores) {
|
||||
steps := liveAssignmentAccountSteps(open)
|
||||
goServing(ctx, func() {
|
||||
runAssignmentWatches(ctx, open.inventory.WatchAssignments, func(a chain.Assignment) {
|
||||
inBackground(ctx, func() {
|
||||
untilSettled(ctx, a.Module+" on "+a.Node, func(ctx context.Context) (string, error) {
|
||||
return steps.settle(ctx, a.Node, a.Module)
|
||||
})
|
||||
})
|
||||
}, func(look []chain.Assignment) {
|
||||
// The start's look, once in the process: one reading of the catalogue, one assignment after another,
|
||||
// then one composition of the bus's node and the grants (issues 1945 and 1277).
|
||||
inBackground(ctx, func() {
|
||||
untilSettled(ctx, "the start's look", func(ctx context.Context) (string, error) {
|
||||
return steps.look(ctx, look)
|
||||
})
|
||||
})
|
||||
}, func() bool {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return false
|
||||
case <-time.After(time.Minute):
|
||||
return true
|
||||
}
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
// runAssignmentWatches watches the assignments again each time a watch ends, until ctx ends or pause says stop: each
|
||||
// assignment that comes to stand is handed to fresh, and the standing ones of the first watch that reads them all (its
|
||||
// Caught mark) to look, once. A watch that ends, or fails to open, before the start's look ran is no start's look: the
|
||||
// next one is, with what it hears from nothing (a fresh assignmentTurns), so no key it handed is dropped and no burst of
|
||||
// fresh settles replaces the look (the review of issue 1945).
|
||||
func runAssignmentWatches(ctx context.Context, watch func(context.Context) (<-chan chain.Entry, error),
|
||||
fresh func(chain.Assignment), look func([]chain.Assignment), pause func() bool) {
|
||||
turns := &assignmentTurns{}
|
||||
first := true
|
||||
for {
|
||||
var atStart []chain.Assignment
|
||||
if err := watchAssignmentsOnce(ctx, watch, turns, first, func(a chain.Assignment, isFresh bool) {
|
||||
if !isFresh {
|
||||
atStart = append(atStart, a)
|
||||
return
|
||||
}
|
||||
fresh(a)
|
||||
}, func() {
|
||||
first = false
|
||||
look(atStart)
|
||||
atStart = nil
|
||||
}); err != nil && ctx.Err() == nil {
|
||||
fmt.Printf("assignments: the assignments are not heard, so a new assignment's bus accounts wait: %v\n", err)
|
||||
}
|
||||
if first {
|
||||
turns = &assignmentTurns{} // the start's look has not run: the next watch hears every standing key anew
|
||||
}
|
||||
if ctx.Err() != nil || !pause() {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// watchAssignmentsOnce watches the assignments until the watch or ctx ends, handing each assignment that comes to stand
|
||||
// to settle. On the process's first watch, the current values are the start's look (not fresh) and read is called once
|
||||
// they are all handed (the chain's Caught mark); every other value, and every value of a watch opened again after one
|
||||
// ended, is fresh: an assignment that came to stand while no watch heard it is settled whole, never as the start's look
|
||||
// (the review of issue 1945).
|
||||
func watchAssignmentsOnce(ctx context.Context, watch func(context.Context) (<-chan chain.Entry, error),
|
||||
turns *assignmentTurns, first bool, settle func(a chain.Assignment, fresh bool), read func()) error {
|
||||
ctx, stop := context.WithCancel(ctx)
|
||||
defer stop()
|
||||
entries, err := watch(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
current := true
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil
|
||||
case e, ok := <-entries:
|
||||
if !ok {
|
||||
if ctx.Err() != nil {
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("the watch ended")
|
||||
}
|
||||
if e.Caught {
|
||||
if current {
|
||||
current = false // the current values are read: what follows is written from now on
|
||||
if first {
|
||||
read()
|
||||
}
|
||||
}
|
||||
continue
|
||||
}
|
||||
var value []byte
|
||||
if !e.Deleted {
|
||||
value = e.Value
|
||||
}
|
||||
if a, turned := turns.heard(e.Key, value); turned {
|
||||
settle(a, !current || !first)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,344 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
"github.com/novox/mesh-controller/internal/chain"
|
||||
)
|
||||
|
||||
// An assignment that comes to stand issues its node's bus accounts and the module's grant (ADR 0543, issue 1945): on
|
||||
// 2026-10-25 at 17:19 mesh-build-node and mesh-build-go, built long before, were assigned to ace, and nothing issued
|
||||
// ace's accounts nor put them in the bus's user list.
|
||||
|
||||
func anAssignmentValue(t *testing.T, a chain.Assignment) []byte {
|
||||
t.Helper()
|
||||
raw, err := json.Marshal(a)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return raw
|
||||
}
|
||||
|
||||
// Only a write that makes the key stand turns it: a rewrite of a standing assignment (the provisioner's plan kept in its
|
||||
// value), an ended one, a pending one or a module only pulled in by another does not. The process's first watch hands
|
||||
// the assignments standing when it starts to the start's look (not fresh), says once that they are all read, and hands
|
||||
// each assignment written afterwards as fresh; a watch opened again after one ended hands every standing assignment it
|
||||
// has not seen stand as fresh and never calls the start's look (the review of issue 1945).
|
||||
func TestAnAssignmentThatComesToStand(t *testing.T) {
|
||||
at := time.Date(2026, 10, 25, 16, 19, 42, 0, time.UTC)
|
||||
stands := chain.Assignment{Node: "ace", Module: "mesh-build-node", Assigned: at}
|
||||
ended := stands
|
||||
ended.Ended = at.Add(time.Hour)
|
||||
pending := stands
|
||||
pending.State = chain.StatePending
|
||||
pulled := chain.Assignment{Node: "ace", Module: "mesh-build-node", RequiredBy: "mesh-build-go"}
|
||||
replanned := stands
|
||||
replanned.Provisioning = json.RawMessage(`{"plan":1}`)
|
||||
|
||||
type write struct {
|
||||
value *chain.Assignment // nil for a delete
|
||||
turning bool
|
||||
}
|
||||
for _, c := range []struct {
|
||||
name string
|
||||
writes []write
|
||||
}{
|
||||
{"a new assignment turns", []write{{&stands, true}}},
|
||||
{"a rewrite of one that stands does not", []write{{&stands, true}, {&replanned, false}, {&stands, false}}},
|
||||
{"an ended one does not", []write{{&ended, false}}},
|
||||
{"assigned again after it ended turns again", []write{{&stands, true}, {&ended, false}, {&stands, true}}},
|
||||
{"a pending one does not, and turns once made", []write{{&pending, false}, {&stands, true}}},
|
||||
{"a module only pulled in does not", []write{{&pulled, false}}},
|
||||
{"one deleted and written again turns again", []write{{&stands, true}, {nil, false}, {&stands, true}}},
|
||||
} {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
turns := &assignmentTurns{}
|
||||
for i, w := range c.writes {
|
||||
var value []byte
|
||||
if w.value != nil {
|
||||
value = anAssignmentValue(t, *w.value)
|
||||
}
|
||||
a, turned := turns.heard("ace.mesh-build-node", value)
|
||||
if turned != w.turning {
|
||||
t.Fatalf("write %d: turned %v, want %v", i, turned, w.turning)
|
||||
}
|
||||
if turned && (a.Node != "ace" || a.Module != "mesh-build-node") {
|
||||
t.Fatalf("write %d: the assignment turned is %s on %s", i, a.Module, a.Node)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
type handed struct {
|
||||
key string
|
||||
fresh bool
|
||||
}
|
||||
for _, first := range []bool{true, false} {
|
||||
t.Run(fmt.Sprintf("the watch, first %v", first), func(t *testing.T) {
|
||||
bucket := chain.NewMemory()
|
||||
put := func(a chain.Assignment) {
|
||||
t.Helper()
|
||||
value := anAssignmentValue(t, a)
|
||||
if e, held, _ := bucket.Get(t.Context(), a.Key()); held {
|
||||
if _, err := bucket.Update(t.Context(), a.Key(), value, e.Revision); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return
|
||||
}
|
||||
if _, err := bucket.Create(t.Context(), a.Key(), value); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
put(chain.Assignment{Node: "g14", Module: "mesh-build-go", Assigned: at})
|
||||
got := make(chan handed, 8)
|
||||
read := make(chan struct{}, 8)
|
||||
ctx, stop := context.WithCancel(t.Context())
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
done <- watchAssignmentsOnce(ctx, bucket.Watch, &assignmentTurns{}, first,
|
||||
func(a chain.Assignment, fresh bool) { got <- handed{a.Key(), fresh} },
|
||||
func() { read <- struct{}{} })
|
||||
}()
|
||||
next := func() handed {
|
||||
t.Helper()
|
||||
select {
|
||||
case h := <-got:
|
||||
return h
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("nothing handed")
|
||||
return handed{}
|
||||
}
|
||||
}
|
||||
// Standing when the watch opened: the start's look on the first watch, fresh on any other.
|
||||
if h := next(); h != (handed{"g14.mesh-build-go", !first}) {
|
||||
t.Fatalf("the standing assignment was handed %+v", h)
|
||||
}
|
||||
put(chain.Assignment{Node: "g14", Module: "mesh-build-go", Assigned: at, Provisioning: json.RawMessage(`{"plan":2}`)})
|
||||
put(chain.Assignment{Node: "ace", Module: "mesh-build-node", Assigned: at})
|
||||
if h := next(); h != (handed{"ace.mesh-build-node", true}) {
|
||||
t.Fatalf("handed %+v, want ace.mesh-build-node as fresh (and nothing for the rewrite before it)", h)
|
||||
}
|
||||
stop()
|
||||
if err := <-done; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if want := map[bool]int{true: 1, false: 0}[first]; len(read) != want {
|
||||
t.Fatalf("the start's look was called %d time(s), want %d", len(read), want)
|
||||
}
|
||||
})
|
||||
}
|
||||
// A first watch that fails to open, or ends before its Caught mark, is no start's look: the next watch is, and hands
|
||||
// every standing assignment to it once, none as fresh (the review of issue 1945: a burst of fresh settles, or keys
|
||||
// handed and dropped).
|
||||
for _, firstWatch := range []string{"fails to open", "ends before its Caught mark"} {
|
||||
t.Run("a first watch that "+firstWatch, func(t *testing.T) {
|
||||
bucket := chain.NewMemory()
|
||||
standing := chain.Assignment{Node: "g14", Module: "mesh-build-go", Assigned: at}
|
||||
if _, err := bucket.Create(t.Context(), standing.Key(), anAssignmentValue(t, standing)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
watches := 0
|
||||
watch := func(ctx context.Context) (<-chan chain.Entry, error) {
|
||||
watches++
|
||||
if watches > 1 {
|
||||
return bucket.Watch(ctx)
|
||||
}
|
||||
if firstWatch == "fails to open" {
|
||||
return nil, errors.New("the chain is not open yet")
|
||||
}
|
||||
cut := make(chan chain.Entry, 1)
|
||||
cut <- chain.Entry{Key: standing.Key(), Value: anAssignmentValue(t, standing)}
|
||||
close(cut)
|
||||
return cut, nil
|
||||
}
|
||||
var looks [][]chain.Assignment
|
||||
var fresh []chain.Assignment
|
||||
ctx, stop := context.WithCancel(t.Context())
|
||||
defer stop()
|
||||
runAssignmentWatches(ctx, watch, func(a chain.Assignment) { fresh = append(fresh, a) },
|
||||
func(look []chain.Assignment) { looks = append(looks, look); stop() },
|
||||
func() bool { return ctx.Err() == nil })
|
||||
if len(looks) != 1 || len(looks[0]) != 1 || looks[0][0].Key() != standing.Key() || len(fresh) != 0 {
|
||||
t.Fatalf("after %d watch(es): looks %v, fresh %v; want one look at %s and nothing fresh",
|
||||
watches, looks, fresh, standing.Key())
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// settle issues what is missing and then puts it on the bus: the node holding the bus composed (its user list) and the
|
||||
// grants written, even when every account was minted before (a module assigned again keeps its account). The start's
|
||||
// look reads the catalogue once, issues for each standing assignment, and composes the bus's node and writes the grants
|
||||
// once when it looked at any. Neither acts for a controller that does not hold the lease, nor for the controller's own
|
||||
// module.
|
||||
func TestSettlingAndTheStartsLook(t *testing.T) {
|
||||
ace := func(m string) chain.Assignment { return chain.Assignment{Node: "ace", Module: m} }
|
||||
cases := []struct {
|
||||
name string
|
||||
look []chain.Assignment // nil: settle mesh-build-node on ace
|
||||
module string
|
||||
notHolding bool
|
||||
issued bool
|
||||
issueErr, composeE error
|
||||
holder string
|
||||
wantIssues int
|
||||
wantComposed []string
|
||||
wantGrants bool
|
||||
wantErr string
|
||||
}{
|
||||
{name: "an account not minted", module: "mesh-build-node", issued: true, holder: "novox", wantIssues: 1,
|
||||
wantComposed: []string{"novox"}, wantGrants: true},
|
||||
{name: "an account minted before: the bus and the grants still", module: "mesh-build-go", holder: "novox",
|
||||
wantIssues: 1, wantComposed: []string{"novox"}, wantGrants: true},
|
||||
{name: "no lease: nothing, and tried again", module: "mesh-build-node", notHolding: true, issued: true,
|
||||
holder: "novox", wantErr: errNotHolding.Error()},
|
||||
{name: "the controller's own module is the terminal's", module: controllerModule, issued: true, holder: "novox"},
|
||||
{name: "no bus holder: the grants still", module: "mesh-build-node", issued: true, wantIssues: 1, wantGrants: true},
|
||||
{name: "an issue that fails is the error, nothing composed", module: "mesh-build-node",
|
||||
issueErr: errors.New("the bus cannot be reached"), holder: "novox", wantIssues: 1, wantErr: "the bus cannot be reached"},
|
||||
{name: "a composition that fails is the error, no grant", module: "mesh-build-node", issued: true, holder: "novox",
|
||||
composeE: errors.New("store refused"), wantIssues: 1, wantComposed: []string{"novox"}, wantErr: "store refused"},
|
||||
{name: "the start's look: each issued, the bus and the grants once",
|
||||
look: []chain.Assignment{ace("mesh-build-node"), ace(controllerModule), ace("mesh-build-go")}, issued: true,
|
||||
holder: "novox", wantIssues: 2, wantComposed: []string{"novox"}, wantGrants: true},
|
||||
{name: "the start's look with every account minted still covers once",
|
||||
look: []chain.Assignment{ace("mesh-build-node"), ace("mesh-build-go")}, holder: "novox", wantIssues: 2,
|
||||
wantComposed: []string{"novox"}, wantGrants: true},
|
||||
{name: "the start's look at nothing does nothing", look: []chain.Assignment{ace(controllerModule)}, holder: "novox"},
|
||||
}
|
||||
for _, c := range cases {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
var composed []string
|
||||
issues, catalogues, grants := 0, 0, 0
|
||||
steps := assignmentAccountSteps{
|
||||
holds: func(context.Context) bool { return !c.notHolding },
|
||||
catalogue: func(context.Context) (map[string]catalogue.Manifest, error) {
|
||||
catalogues++
|
||||
return map[string]catalogue.Manifest{}, nil
|
||||
},
|
||||
issue: func(context.Context, map[string]catalogue.Manifest, string, string) (bool, string, error) {
|
||||
issues++
|
||||
if c.issued {
|
||||
return true, "its bus credential issued", c.issueErr
|
||||
}
|
||||
return false, "", c.issueErr
|
||||
},
|
||||
busHolder: func(context.Context) (string, error) { return c.holder, nil },
|
||||
compose: func(_ context.Context, nodes []string) error {
|
||||
composed = append(composed, nodes...)
|
||||
return c.composeE
|
||||
},
|
||||
grants: func(context.Context) error { grants++; return nil },
|
||||
}
|
||||
var said string
|
||||
var err error
|
||||
if c.look != nil {
|
||||
said, err = steps.look(t.Context(), c.look)
|
||||
} else {
|
||||
said, err = steps.settle(t.Context(), "ace", c.module)
|
||||
}
|
||||
if c.wantErr != "" {
|
||||
if err == nil || !strings.Contains(err.Error(), c.wantErr) {
|
||||
t.Fatalf("error %v, want one saying %q", err, c.wantErr)
|
||||
}
|
||||
} else if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if issues != c.wantIssues || catalogues > 1 {
|
||||
t.Fatalf("%d issue(s) from %d reading(s) of the catalogue, want %d from at most one", issues, catalogues, c.wantIssues)
|
||||
}
|
||||
if strings.Join(composed, ",") != strings.Join(c.wantComposed, ",") {
|
||||
t.Fatalf("composed %v, want %v", composed, c.wantComposed)
|
||||
}
|
||||
if want := map[bool]int{true: 1}[c.wantGrants]; grants != want {
|
||||
t.Fatalf("grants written %d time(s), want %d", grants, want)
|
||||
}
|
||||
if c.wantGrants && c.wantErr == "" && !strings.Contains(said, "the grants written") {
|
||||
t.Fatalf("said %q, want it to name the grants", said)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// The accounts a module speaks with on a node: the module's own when it reads `broker`, its provisioner client's when
|
||||
// it runs one alone, none when its accounts are its processes' or it does not speak on the bus. A process's account is
|
||||
// held in the provisioner client's form, named for the process and carrying no claims, and may be issued only when
|
||||
// the process declares its own secret broker-<process id> (ADR 0436 §2; issue 1181).
|
||||
func TestTheAccountsAModuleSpeaksWithOnANode(t *testing.T) {
|
||||
client := aProvisioner()
|
||||
delete(client.OwnSecrets, "broker")
|
||||
for _, c := range []struct {
|
||||
name string
|
||||
m catalogue.Manifest
|
||||
user string
|
||||
own bool
|
||||
}{
|
||||
{"a module reading broker", catalogue.Manifest{Module: "mesh-build-node",
|
||||
OwnSecrets: catalogue.OwnSecrets{"broker": {Path: "/var/lib/mesh-build-node/broker"}}},
|
||||
broker.Principal{Kind: broker.KindModule, Node: "ace", Module: "mesh-build-node"}.Username(), true},
|
||||
{"a module reading broker and running the client", aProvisioner(),
|
||||
broker.Principal{Kind: broker.KindModule, Node: "ace", Module: "mesh-provisioner"}.Username(), true},
|
||||
{"the provisioner client alone", client,
|
||||
broker.Principal{Kind: broker.KindProvisioner, Node: "ace", Module: "mesh-provisioner"}.Username(), true},
|
||||
{"no broker, no client", catalogue.Manifest{Module: "hello-web"}, "", false},
|
||||
} {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
user, own := ownAccountOf(c.m, "ace")
|
||||
if user != c.user || own != c.own {
|
||||
t.Fatalf("got %q %v, want %q %v", user, own, c.user, c.own)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
t.Run("a process's account", func(t *testing.T) {
|
||||
user := broker.Principal{Kind: broker.KindProcess, Node: "novox", Module: "mesh-mail", Process: "send"}.Username()
|
||||
if user != "novox.mesh-mail.send" {
|
||||
t.Fatalf("the process's username: %q", user)
|
||||
}
|
||||
held, err := heldProcessAccount(broker.Broker{Fingerprint: "abc"}, "novox.example:4222", "novox", "mesh-mail",
|
||||
user, "a-password")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var got map[string]any
|
||||
if err := json.Unmarshal(held, &got); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
want := map[string]any{"url": "nats://novox.example:4222", "fingerprint": "abc", "node": "novox",
|
||||
"module": "mesh-mail", "user": "novox.mesh-mail.send", "password": "a-password"}
|
||||
for k, v := range want {
|
||||
if got[k] != v {
|
||||
t.Errorf("%s: %v, want %v", k, got[k], v)
|
||||
}
|
||||
}
|
||||
if len(got) != len(want) {
|
||||
t.Errorf("the held account carries %v; want exactly %v (never claims — a process holds no seat)", got, want)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("a process may be issued only with its own broker secret", func(t *testing.T) {
|
||||
// mesh-mail's real shape: no `broker` of its own, its process declaring bus (review round 1 of issue 1181).
|
||||
if err := mayIssue(mailSenderWithProcess()); err != nil {
|
||||
t.Fatalf("a module with no broker of its own, whose process declares bus, was refused: %v", err)
|
||||
}
|
||||
m := mailSenderWithProcess()
|
||||
delete(m.OwnSecrets, "broker-send")
|
||||
err := mayIssue(m)
|
||||
if err == nil {
|
||||
t.Fatal("a process with no broker-<id> own secret was issued an account")
|
||||
}
|
||||
for _, want := range []string{"mesh-mail", "send", "broker-send", "own-secrets"} {
|
||||
if !strings.Contains(err.Error(), want) {
|
||||
t.Errorf("the refusal does not say %q: %v", want, err)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -49,7 +49,7 @@ func attachTheChain(ctx context.Context, inv *inventory.Inventory, server *link.
|
||||
// keys laid out before skips them, so this takes seconds; one that fails is tried again beside the serving, loudly.
|
||||
if err := inv.LayBuildsOut(ctx, func(format string, args ...any) { fmt.Printf("chain: "+format+"\n", args...) }); err != nil {
|
||||
fmt.Printf("chain: the builds are not laid out under their modules yet, and none is read until they are: %v\n", err)
|
||||
goServing(ctx, func() { layBuildsOutSoon(ctx, inv) })
|
||||
go layBuildsOutSoon(ctx, inv)
|
||||
}
|
||||
// The requests the move of delivery 1 found never handed over are not asked again: the controller asks no build
|
||||
// (novox/hq ADR 0501; issue 1851, A10), and the builder builds every fingerprint the indexer writes.
|
||||
@@ -80,11 +80,7 @@ func layBuildsOutSoon(ctx context.Context, inv *inventory.Inventory) {
|
||||
Source: "chain", Summary: "the builds are not laid out under their modules, so none is read: "}
|
||||
if err != nil {
|
||||
fmt.Printf("chain: the builds are not laid out under their modules, and none is read until they are: %v\n", err)
|
||||
if ctx.Err() != nil {
|
||||
return // stopped: the next controller lays them out
|
||||
}
|
||||
// A cancelled layout is tried again and raises nothing: a cancellation is no fault of the chain (issue 1945).
|
||||
if keeper := conditionsFrom; keeper != nil && !failed && !errors.Is(err, context.Canceled) {
|
||||
if keeper := conditionsFrom; keeper != nil && !failed {
|
||||
o.Summary += err.Error()
|
||||
if _, oerr := keeper.Observe(ctx, o); oerr == nil {
|
||||
failed = true
|
||||
|
||||
@@ -326,13 +326,9 @@ func restOfTheMesh(ctx context.Context, open *stores, shelf map[string]catalogue
|
||||
return world, nil
|
||||
}
|
||||
|
||||
// restOfTheMeshOf is restOfTheMesh for a caller that holds the inventory alone: on the serving controller's stores where it
|
||||
// lends them (sharedStores), never a pool per call; elsewhere (a verb run as its own process) the other stores the Input
|
||||
// reads are opened for the call and let go of again.
|
||||
// restOfTheMeshOf is restOfTheMesh for a caller that holds the inventory alone: the other stores the Input reads are opened for
|
||||
// the call and let go of again.
|
||||
func restOfTheMeshOf(ctx context.Context, inv *inventory.Inventory, shelf map[string]catalogue.Manifest, exclude string) (catalogue.World, error) {
|
||||
if shared, ok := sharedStoresOf(inv); ok {
|
||||
return restOfTheMeshOn(ctx, shared, shelf, exclude)
|
||||
}
|
||||
open := &stores{inventory: inv}
|
||||
defer func() {
|
||||
if open.licences != nil {
|
||||
@@ -342,12 +338,9 @@ func restOfTheMeshOf(ctx context.Context, inv *inventory.Inventory, shelf map[st
|
||||
open.identity.Close()
|
||||
}
|
||||
}()
|
||||
return restOfTheMeshOn(ctx, open, shelf, exclude)
|
||||
return restOfTheMesh(ctx, open, shelf, exclude)
|
||||
}
|
||||
|
||||
// restOfTheMeshOn is restOfTheMesh; a variable so a test reads which stores it is handed.
|
||||
var restOfTheMeshOn = restOfTheMesh
|
||||
|
||||
// sayFound says the lines a composition found while composing, once each: a machine left out of the rest of the mesh, a seat
|
||||
// that two claimants could take, the bus's users left out for want of a credential.
|
||||
func sayFound(ctx context.Context, d compose.Declared) {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -140,35 +140,4 @@ func TestTheDeclarationsStreamGoesOnceEveryNodeRunsTheNewEngine(t *testing.T) {
|
||||
if !declarationsRetiredNow(ctx, m.open.inventory) {
|
||||
t.Errorf("a raise after the retirement would assert the stream again")
|
||||
}
|
||||
|
||||
// Issue 1945: a node-engine other than the registered one landing on laptop (a newer build on its way) leaves the
|
||||
// declarations retired, so what the health check, the probes and healer H3 expect of the bus names neither the
|
||||
// stream nor a node's consumer on it, and asserting the bus's objects again does not make the stream anew.
|
||||
have("laptop", fpOf('7'))
|
||||
if !declarationsRetiredNow(ctx, m.open.inventory) {
|
||||
t.Fatalf("with laptop on another node-engine after the stream went, the declarations are no longer retired")
|
||||
}
|
||||
streams, consumers, err := expectedBusObjects(ctx, m.open.inventory)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, s := range streams {
|
||||
if s.Name == broker.DeclarationsStream {
|
||||
t.Errorf("the bus's health check and probe D7 still expect the stream %s", s.Name)
|
||||
}
|
||||
}
|
||||
for _, c := range consumers {
|
||||
if c.Stream == broker.DeclarationsStream {
|
||||
t.Errorf("the bus's health check and probe D6 still expect %s on %s", c.Name, c.Stream)
|
||||
}
|
||||
}
|
||||
bus, err := broker.Dial(m.conn.ConnectedUrl())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(bus.Close)
|
||||
_ = assertOnSend(ctx, m.open.inventory, bus, " ")
|
||||
if _, err := m.js.Stream(ctx, broker.DeclarationsStream); !errors.Is(err, jetstream.ErrStreamNotFound) {
|
||||
t.Errorf("healer H3's assertion of the bus's objects made the stream anew: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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,45 +129,19 @@ 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 := takeOnceInSlot(ctx, open, module)
|
||||
err := doneKeyTakeOnce(ctx, open, module)
|
||||
o := doneKeyObservation(module)
|
||||
if err == nil {
|
||||
// Cleared on every success, whoever raised it: a controller before a restart or another follower (issue
|
||||
// 1945). A condition that is not open costs one read.
|
||||
if keeper := conditionsFrom; keeper != 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
|
||||
}
|
||||
if errors.Is(err, errKeyWaits) {
|
||||
@@ -196,12 +163,7 @@ func followDoneKey(ctx context.Context, open *stores, module string) {
|
||||
}
|
||||
continue
|
||||
}
|
||||
if ctx.Err() != nil {
|
||||
return // stopped: a take-in cut by the stop is no fault of the build, and the next controller takes it in
|
||||
}
|
||||
if keeper := conditionsFrom; keeper != nil && !failed && !errors.Is(err, context.Canceled) {
|
||||
// A take-in whose own step was cancelled is tried again, and raises nothing: a cancellation is no fault
|
||||
// of the build (issue 1945: 30 urgent conditions live, each a take-in cut by a restart).
|
||||
if keeper := conditionsFrom; keeper != nil && !failed {
|
||||
o.Summary += err.Error()
|
||||
if _, oerr := keeper.Observe(ctx, o); oerr == nil {
|
||||
failed = true
|
||||
@@ -289,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: ")
|
||||
},
|
||||
}
|
||||
}
|
||||
@@ -309,18 +279,18 @@ func watchDoneKeys(ctx context.Context, open *stores) {
|
||||
return
|
||||
}
|
||||
if err := builderkeys.Watch(ctx, keys.JS, func(module string) {
|
||||
inBackground(ctx, func() { takeDoneKey(ctx, open, module) })
|
||||
inBackground(func() { takeDoneKey(ctx, open, module) })
|
||||
}); err != nil {
|
||||
closeKeys()
|
||||
fmt.Printf("done keys: the builder's keys are not watched: %v\n", err)
|
||||
return
|
||||
}
|
||||
goServing(ctx, func() {
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
closeKeys()
|
||||
})
|
||||
}()
|
||||
// And every module once now, after the watch is open so no write falls between the two.
|
||||
inBackground(ctx, func() {
|
||||
inBackground(func() {
|
||||
doneKeysAtStart(ctx, catalogedModules(open), func(module string) { takeDoneKeyAtStart(ctx, open, module) })
|
||||
})
|
||||
}
|
||||
@@ -337,61 +307,24 @@ 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, and how many run at once in all.
|
||||
const startTakeIns = 16
|
||||
|
||||
// takeInSlots are the take-ins that may run at once in this process, the watch's and the start's look's together: a
|
||||
// burst of done keys (about 120 at 16:07:36 on 2026-10-25) once started a take-in each, all at once, and their reads
|
||||
// filled the store until it refused every connection (issues 1945 and 1277).
|
||||
var takeInSlots = make(chan struct{}, startTakeIns)
|
||||
|
||||
// takeOnceInSlot is one take-in once a slot is free; a take-in that waits for its next try holds none.
|
||||
func takeOnceInSlot(ctx context.Context, open *stores, module string) error {
|
||||
select {
|
||||
case takeInSlots <- struct{}{}:
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
defer func() { <-takeInSlots }()
|
||||
return doneKeyTakeOnce(ctx, open, module)
|
||||
}
|
||||
|
||||
// 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 := takeOnceInSlot(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.
|
||||
inBackground(ctx, func() { takeDoneKey(ctx, open, module) })
|
||||
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) })
|
||||
}
|
||||
}
|
||||
|
||||
@@ -409,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
|
||||
}
|
||||
|
||||
@@ -1,83 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/conditions"
|
||||
)
|
||||
|
||||
// A take-in cut by a cancellation raises no condition (issue 1945: on 2026-10-25 thirty urgent fingerprint-not-composed
|
||||
// conditions were live, each a take-in a stop or a restart cut with "context canceled"): when the controller stops, the
|
||||
// next one takes it in; when a step of its own was cancelled, it is tried again.
|
||||
func TestATakeInCutByACancellationRaisesNothing(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
stop bool // the controller's context ends, else only the step's error is a cancellation
|
||||
}{{"the controller stops", true}, {"a step was cancelled", false}} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
was, wasRetry, wasSay, wasKeeper := doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom
|
||||
t.Cleanup(func() {
|
||||
doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom = was, wasRetry, wasSay, wasKeeper
|
||||
})
|
||||
store := conditions.NewInMemory()
|
||||
k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store,
|
||||
Teller: &conditions.Told{}})
|
||||
t.Cleanup(func() { k.Close(context.Background()) })
|
||||
conditionsFrom = k
|
||||
doneKeyRetry = time.Millisecond
|
||||
doneKeysSay = func(string) {}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
calls := 0
|
||||
doneKeyTakeOnce = func(context.Context, *stores, string) error {
|
||||
calls++
|
||||
if tc.stop {
|
||||
cancel() // stopped mid-way
|
||||
return fmt.Errorf("the module cannot be composed: %w", context.Canceled)
|
||||
}
|
||||
if calls == 1 {
|
||||
return fmt.Errorf("the module cannot be composed: %w", context.Canceled)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
done := make(chan struct{})
|
||||
go func() { takeDoneKey(ctx, nil, "cut-module"); close(done) }()
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("the take-in did not end")
|
||||
}
|
||||
if !tc.stop && calls != 2 {
|
||||
t.Errorf("a take-in whose step was cancelled was tried %d time(s), want 2", calls)
|
||||
}
|
||||
if c, found, err := conditions.ReadOne(t.Context(), store, doneKeyObservation("cut-module").Key()); err != nil || found {
|
||||
t.Fatalf("a take-in cut by a cancellation raised %+v (found %v, %v)", c, found, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// A take-in that succeeds clears the module's fingerprint-not-composed condition whoever raised it: a controller before a
|
||||
// restart, or another follower (issue 1945: the operator read "a build needs a look" for take-ins that had since worked).
|
||||
func TestASucceedingTakeInClearsAConditionAnotherRaised(t *testing.T) {
|
||||
was, wasRetry, wasSay, wasKeeper := doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom
|
||||
t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom = was, wasRetry, wasSay, wasKeeper })
|
||||
store := conditions.NewInMemory()
|
||||
k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store, Teller: &conditions.Told{}})
|
||||
t.Cleanup(func() { k.Close(context.Background()) })
|
||||
conditionsFrom = k
|
||||
doneKeysSay = func(string) {}
|
||||
o := doneKeyObservation("raised-before")
|
||||
o.Summary += "the store refused the connection"
|
||||
if _, err := k.Observe(t.Context(), o); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
doneKeyTakeOnce = func(context.Context, *stores, string) error { return nil }
|
||||
takeDoneKey(t.Context(), nil, "raised-before")
|
||||
if c, found, err := conditions.ReadOne(t.Context(), store, o.Key()); err != nil || found {
|
||||
t.Fatalf("a take-in that succeeded left the condition another raised: %+v (found %v, %v)", c, found, err)
|
||||
}
|
||||
}
|
||||
@@ -1,58 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// A burst of done keys runs at most startTakeIns take-ins at once, the watch's and the start's look together: at
|
||||
// 16:07:36 on 2026-10-25 the builder rewrote about 120 done keys at once, each woke its own take-in, and each opened
|
||||
// pools of its own until novox's store refused every connection (issues 1945 and 1277).
|
||||
func TestABurstOfDoneKeysRunsAtMostSixteenTakeInsAtOnce(t *testing.T) {
|
||||
was := doneKeyTakeOnce
|
||||
t.Cleanup(func() { doneKeyTakeOnce = was })
|
||||
var running, most atomic.Int64
|
||||
var taken atomic.Int64
|
||||
doneKeyTakeOnce = func(context.Context, *stores, string) error {
|
||||
now := running.Add(1)
|
||||
for {
|
||||
m := most.Load()
|
||||
if now <= m || most.CompareAndSwap(m, now) {
|
||||
break
|
||||
}
|
||||
}
|
||||
time.Sleep(2 * time.Millisecond)
|
||||
running.Add(-1)
|
||||
taken.Add(1)
|
||||
return nil
|
||||
}
|
||||
ctx := context.Background()
|
||||
var wg sync.WaitGroup
|
||||
// The watch's wakes: one take-in each, in the background, as watchDoneKeys starts them.
|
||||
for i := range 120 {
|
||||
wg.Add(1)
|
||||
go func() { defer wg.Done(); takeDoneKey(ctx, nil, fmt.Sprintf("burst-%03d", i)) }()
|
||||
}
|
||||
// And the start's look at the same moment, over other modules.
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
var start []string
|
||||
for i := range 40 {
|
||||
start = append(start, fmt.Sprintf("start-%03d", i))
|
||||
}
|
||||
doneKeysAtStart(ctx, func(context.Context) ([]string, error) { return start, nil },
|
||||
func(m string) { takeDoneKeyAtStart(ctx, nil, m) })
|
||||
}()
|
||||
wg.Wait()
|
||||
if got := taken.Load(); got != 160 {
|
||||
t.Fatalf("took in %d modules, want 160", got)
|
||||
}
|
||||
if m := most.Load(); m > startTakeIns {
|
||||
t.Fatalf("ran %d take-ins at once, want at most %d", m, startTakeIns)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -134,7 +134,7 @@ func watchHavesForJudgement(ctx context.Context, open *stores) {
|
||||
fmt.Printf("compositions: the haves are not watched: %v\n", err)
|
||||
return
|
||||
}
|
||||
goServing(ctx, func() {
|
||||
go func() {
|
||||
defer bus.Close()
|
||||
defer w.Stop()
|
||||
for {
|
||||
@@ -152,7 +152,7 @@ func watchHavesForJudgement(ctx context.Context, open *stores) {
|
||||
if err != nil || isNode {
|
||||
continue
|
||||
}
|
||||
inBackground(ctx, func() {
|
||||
inBackground(func() {
|
||||
if err := rejudgeOn(ctx, open, []string{node}, module); err != nil {
|
||||
fmt.Printf("compositions: %s on %s is not judged again: %v\n", module, node, err)
|
||||
}
|
||||
@@ -165,5 +165,5 @@ func watchHavesForJudgement(ctx context.Context, open *stores) {
|
||||
})
|
||||
}
|
||||
}
|
||||
})
|
||||
}()
|
||||
}
|
||||
|
||||
@@ -608,11 +608,8 @@ func keepWhatServeSets(t *testing.T) {
|
||||
t.Setenv(servedVar, "")
|
||||
unheld, status, given, check, follow, keeper, hand := logUnheldChanges, statusFrom, givenEvents, checkEvents,
|
||||
link.Calls.Follow, conditionsFrom, handActConn
|
||||
doctor, asker := doctorFrom, askerFrom.Load()
|
||||
t.Cleanup(func() {
|
||||
logUnheldChanges, statusFrom, givenEvents, checkEvents, link.Calls.Follow, conditionsFrom, handActConn =
|
||||
unheld, status, given, check, follow, keeper, hand
|
||||
doctorFrom = doctor
|
||||
askerFrom.Store(asker)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -135,18 +135,9 @@ func declarationsDone(nodes []string, runs map[string]string, current string) bo
|
||||
return true
|
||||
}
|
||||
|
||||
// declarationsRetiredOn reads whether the declarations are done on the mesh now: their stream gone from the bus, or
|
||||
// every node's have of the node-engine against the fingerprint it is registered at.
|
||||
//
|
||||
// **A stream that is gone stays gone** (issue 1945). Release 6 sends no declaration (ADR 0539 clauses 5 and 11), so a
|
||||
// stream made again would hold no node's last declaration and help no node; and the done test alone is not lasting:
|
||||
// a node-engine newer than the one registered landing, or a node asleep on an older have, makes it false again after
|
||||
// the start deleted the stream, and then the raise, the bus's health check (H-bus), the probes (D6, D7) and healer H3
|
||||
// expected the stream and every node's consumer again, raised them missing and made them anew (2026-10-25 14:52).
|
||||
// declarationsRetiredOn reads whether the declarations are done on the mesh now: every node's have of the node-engine
|
||||
// against the fingerprint it is registered at.
|
||||
func declarationsRetiredOn(ctx context.Context, inv *inventory.Inventory, js jetstream.JetStream) (bool, error) {
|
||||
if _, err := js.Stream(ctx, broker.DeclarationsStream); errors.Is(err, jetstream.ErrStreamNotFound) {
|
||||
return true, nil
|
||||
}
|
||||
fps, err := currentFingerprints(ctx, inv)
|
||||
if err != nil {
|
||||
return false, err
|
||||
@@ -171,8 +162,7 @@ func declarationsRetiredOn(ctx context.Context, inv *inventory.Inventory, js jet
|
||||
}
|
||||
|
||||
// retireDeclarations deletes the declarations' stream, with every node's consumer on it, once every node runs the new
|
||||
// node-engine, and says it once; a stream already gone is said by nobody, and is done. It answers whether the
|
||||
// declarations are done.
|
||||
// node-engine, and says it once; a stream already gone is said by nobody. It answers whether the declarations are done.
|
||||
func retireDeclarations(ctx context.Context, open *stores) (bool, error) {
|
||||
if _, err := theLease.epoch(ctx); err != nil {
|
||||
return false, err
|
||||
@@ -334,9 +324,8 @@ func replacedComposition(in compose.Input, r replacedController) (sdkdelivery.Co
|
||||
return c, nil
|
||||
}
|
||||
|
||||
// declarationsRetiredNow says the declarations are done on the mesh now, for a raise of the bus's objects and for what
|
||||
// the health check, the probes and the healers expect of the bus; a bus or a store it cannot read is not done, so a
|
||||
// stream still there is asserted as before.
|
||||
// declarationsRetiredNow says the declarations are done on the mesh now, for a raise of the bus's objects; a bus or a
|
||||
// store it cannot read is not done, so the stream is asserted as before.
|
||||
func declarationsRetiredNow(ctx context.Context, inv *inventory.Inventory) bool {
|
||||
bus, err := aBus()
|
||||
if err != nil {
|
||||
|
||||
@@ -3,7 +3,6 @@ package main
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
@@ -111,28 +110,3 @@ func TestTheDeclarationsGoOnlyOnceEveryNodeRunsTheNewEngine(t *testing.T) {
|
||||
t.Errorf("done with no fingerprint registered, or no node")
|
||||
}
|
||||
}
|
||||
|
||||
// streamsGone is a bus that holds none of the streams asked for; every other part of it is unread.
|
||||
type streamsGone struct{ jetstream.JetStream }
|
||||
|
||||
func (streamsGone) Stream(context.Context, string) (jetstream.Stream, error) {
|
||||
return nil, jetstream.ErrStreamNotFound
|
||||
}
|
||||
|
||||
// Issue 1945 (the alarm of a stream the start deleted): once the declarations' stream is gone from the bus, the
|
||||
// declarations are retired whatever the nodes run now, so no raise, health check, probe or healer expects the stream
|
||||
// or a node's consumer on it again. A node-engine newer than the registered one, or a node asleep on an older have,
|
||||
// made the done test false after the stream went, and the bus's objects named it missing again (2026-10-25 14:52).
|
||||
func TestTheDeclarationsStayRetiredOnceTheirStreamIsGone(t *testing.T) {
|
||||
retired, err := func() (retired bool, err error) {
|
||||
defer func() {
|
||||
if p := recover(); p != nil {
|
||||
err = fmt.Errorf("it read what the nodes run although the stream is gone: %v", p)
|
||||
}
|
||||
}()
|
||||
return declarationsRetiredOn(t.Context(), nil, streamsGone{})
|
||||
}()
|
||||
if err != nil || !retired {
|
||||
t.Fatalf("with no declarations' stream on the bus: retired %v (%v), want retired", retired, err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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...)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -49,7 +49,7 @@ func keeperOn(ctx context.Context, conn *nats.Conn, looks conditions.LookStore)
|
||||
Say: func(format string, args ...any) { fmt.Fprintf(os.Stderr, format+"\n", args...) },
|
||||
// What status leads with changed: composed again soon (a nudge outside the serving controller
|
||||
// does nothing).
|
||||
Changed: func() { statusFrom.nudge(); askerFrom.Load().nudge() },
|
||||
Changed: func() { statusFrom.nudge(); askerFrom.nudge() },
|
||||
// Written under the lease, carrying its epoch (novox/hq to-be 45 §6).
|
||||
Epoch: func() (uint64, error) { return theLease.epoch(context.WithoutCancel(ctx)) }}), nil
|
||||
}
|
||||
|
||||
@@ -225,9 +225,6 @@ func filesNotPutObservation(node, summary string) conditions.Observation {
|
||||
// notePut records how putting a send's objects went: a failure counts, the second in a row raises the condition, a success
|
||||
// clears the count and the condition.
|
||||
func notePut(ctx context.Context, node string, put error) {
|
||||
if errors.Is(put, context.Canceled) || ctx.Err() != nil {
|
||||
return // a send cut by a cancellation or a stop says nothing of the bucket (issue 1945)
|
||||
}
|
||||
filesFailMu.Lock()
|
||||
if put == nil {
|
||||
delete(filesFailed, node)
|
||||
|
||||
@@ -415,7 +415,6 @@ func TestAPutThatFailsSendAfterSendRaisesAConditionAndClearsWhenItWorks(t *testi
|
||||
was := conditionsFrom
|
||||
conditionsFrom = k
|
||||
t.Cleanup(func() { conditionsFrom = was })
|
||||
forgetPutFailures(t, "cond-node")
|
||||
key := filesNotPutObservation("cond-node", "").Key()
|
||||
open := func() bool {
|
||||
_, found, err := conditions.ReadOne(t.Context(), store, key)
|
||||
@@ -443,36 +442,6 @@ func TestAPutThatFailsSendAfterSendRaisesAConditionAndClearsWhenItWorks(t *testi
|
||||
}
|
||||
}
|
||||
|
||||
// A send cut by a cancellation counts as no failure to put its files, however often (issue 1945: a cancellation is no fault
|
||||
// of the bucket).
|
||||
func TestAPutCutByACancellationRaisesNothing(t *testing.T) {
|
||||
store := conditions.NewInMemory()
|
||||
k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store, Teller: &conditions.Told{}})
|
||||
t.Cleanup(func() { k.Close(context.Background()) })
|
||||
was := conditionsFrom
|
||||
conditionsFrom = k
|
||||
t.Cleanup(func() { conditionsFrom = was })
|
||||
forgetPutFailures(t, "cut-node")
|
||||
cut := fmt.Errorf("the files could not be put: %w", context.Canceled)
|
||||
for range filesNotPutAfter + 1 {
|
||||
notePut(t.Context(), "cut-node", cut)
|
||||
}
|
||||
if _, found, err := conditions.ReadOne(t.Context(), store, filesNotPutObservation("cut-node", "").Key()); err != nil || found {
|
||||
t.Fatalf("sends cut by a cancellation raised the condition (found %v, %v)", found, err)
|
||||
}
|
||||
}
|
||||
|
||||
// forgetPutFailures starts the node's count of failed puts at nothing and leaves it so, as a run of the test again finds it.
|
||||
func forgetPutFailures(t *testing.T, node string) {
|
||||
forget := func() {
|
||||
filesFailMu.Lock()
|
||||
delete(filesFailed, node)
|
||||
filesFailMu.Unlock()
|
||||
}
|
||||
forget()
|
||||
t.Cleanup(forget)
|
||||
}
|
||||
|
||||
// aBusThatSays is a bus that keeps the declarations published to it and, at each, asks whether the node's bucket is held.
|
||||
type aBusThatSays struct {
|
||||
link.Bus
|
||||
|
||||
@@ -87,29 +87,3 @@ func TestTheDeclaredPackagesMirrorNamesWhatEachNodesModulesDeclare(t *testing.T)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 1945/AC-mirror-5: the digest the declared-packages mirror skips an unchanged pass by is made from a real reading of the
|
||||
// stores, is the same over two readings of the same records, and moves when an assignment does.
|
||||
func TestTheInputDigestOfTheStoresIsStable(t *testing.T) {
|
||||
open := aMesh(t)
|
||||
ctx := t.Context()
|
||||
register(t, open, catalogue.Manifest{Module: "viewer", Version: "1", Resources: []map[string]any{
|
||||
{"id": "pkg-curl", "type": "package", "package": "curl"}}})
|
||||
read := func() string {
|
||||
in, err := readComposeInput(ctx, open)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return inputDigest(in)
|
||||
}
|
||||
first, second := read(), read()
|
||||
if first == "" || first != second {
|
||||
t.Fatalf("two readings of the same records gave the digests %q and %q", first, second)
|
||||
}
|
||||
if _, err := assign(ctx, open, "anchor", "viewer"); err != nil {
|
||||
t.Fatalf("assign: %v", err)
|
||||
}
|
||||
if read() == first {
|
||||
t.Errorf("an assignment left the digest of the stores' Input at %s", first)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
@@ -169,7 +120,7 @@ func watchHavesForGiven(ctx context.Context, open *stores) {
|
||||
log.Printf("given values: the haves are not watched: %v", err)
|
||||
return
|
||||
}
|
||||
goServing(ctx, func() {
|
||||
go func() {
|
||||
defer bus.Close()
|
||||
defer w.Stop()
|
||||
for {
|
||||
@@ -195,8 +146,8 @@ func watchHavesForGiven(ctx context.Context, open *stores) {
|
||||
if at.IsZero() {
|
||||
continue
|
||||
}
|
||||
inBackground(ctx, func() { replaceGiven(ctx, open, node, map[string]time.Time{module: at}) })
|
||||
inBackground(func() { replaceGiven(ctx, open, node, map[string]time.Time{module: at}) })
|
||||
}
|
||||
}
|
||||
})
|
||||
}()
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -965,26 +965,6 @@ func processPrincipalFor(m catalogue.Manifest, processID string) (broker.Process
|
||||
return broker.ProcessPrincipal{}, false
|
||||
}
|
||||
|
||||
// mintOnlyIfAbsent marks a context whose issue mints only accounts no row names yet (onlyIfAbsent).
|
||||
type mintOnlyIfAbsent struct{}
|
||||
|
||||
// onlyIfAbsent is ctx for an issue that must never replace a minted account: the assign act's and the assignments'
|
||||
// watch's, which may run at once for one account (novox/hq ADR 0543, the review of issue 1945).
|
||||
func onlyIfAbsent(ctx context.Context) context.Context {
|
||||
return context.WithValue(ctx, mintOnlyIfAbsent{}, true)
|
||||
}
|
||||
|
||||
// mintBusUser mints u's password. By default it replaces whatever stands (a rotation: `module issue`); under
|
||||
// onlyIfAbsent it mints only when no row names u, and won false says another caller minted it first and seals it,
|
||||
// so this one seals nothing.
|
||||
func mintBusUser(ctx context.Context, inv *inventory.Inventory, u inventory.BusUser) (string, bool, error) {
|
||||
if only, _ := ctx.Value(mintOnlyIfAbsent{}).(bool); only {
|
||||
return inv.MintBusPasswordIfAbsent(ctx, u)
|
||||
}
|
||||
password, err := inv.MintBusPassword(ctx, u)
|
||||
return password, err == nil, err
|
||||
}
|
||||
|
||||
// issueOnTheNewBus gives an assigned module its credential on the bus being built.
|
||||
//
|
||||
// **Three things differ from a management call, and each is the point of the move.** The credential
|
||||
@@ -1010,16 +990,14 @@ func issueOnTheNewBus(ctx context.Context, inv *inventory.Inventory, m catalogue
|
||||
}
|
||||
if _, reads := m.OwnSecrets["broker"]; reads {
|
||||
user := broker.Principal{Kind: broker.KindModule, Node: node, Module: m.Module}.Username()
|
||||
password, won, err := mintBusUser(ctx, inv, inventory.BusUser{
|
||||
password, err := inv.MintBusPassword(ctx, inventory.BusUser{
|
||||
Username: user, Kind: busKindOf(m.Module), Node: node, Module: m.Module,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if won {
|
||||
if err := issueWith(ctx, inv, m, node, busAddress, known, reachable, user, password); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := issueWith(ctx, inv, m, node, busAddress, known, reachable, user, password); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
// Every process with a bus declaration gets an account of its own (novox/hq ADR 0436 §1, §2), minted and
|
||||
@@ -1043,11 +1021,11 @@ func issueOnTheNewBus(ctx context.Context, inv *inventory.Inventory, m catalogue
|
||||
func issueProcess(ctx context.Context, inv *inventory.Inventory, m catalogue.Manifest, pr broker.ProcessPrincipal,
|
||||
node, busAddress string, known broker.Broker, reachable string) error {
|
||||
user := broker.Principal{Kind: broker.KindProcess, Node: node, Module: m.Module, Process: pr.Process}.Username()
|
||||
password, won, err := mintBusUser(ctx, inv, inventory.BusUser{
|
||||
password, err := inv.MintBusPassword(ctx, inventory.BusUser{
|
||||
Username: user, Kind: inventory.BusProcess, Node: node, Module: m.Module,
|
||||
})
|
||||
if err != nil || !won {
|
||||
return err // lost: the caller that minted it seals it
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
name := processSecretName(pr.Process)
|
||||
held, err := heldProcessAccount(known, reachable, node, m.Module, user, password)
|
||||
@@ -1113,11 +1091,11 @@ func runsTheProvisionerClient(m catalogue.Manifest) bool {
|
||||
func issueProvisionerClient(ctx context.Context, inv *inventory.Inventory, m catalogue.Manifest, node string,
|
||||
known broker.Broker, reachable string) error {
|
||||
user := broker.Principal{Kind: broker.KindProvisioner, Node: node, Module: m.Module}.Username()
|
||||
password, won, err := mintBusUser(ctx, inv, inventory.BusUser{
|
||||
password, err := inv.MintBusPassword(ctx, inventory.BusUser{
|
||||
Username: user, Kind: inventory.BusProvisioner, Node: node, Module: m.Module,
|
||||
})
|
||||
if err != nil || !won {
|
||||
return err // lost: the caller that minted it seals it
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
held, err := json.Marshal(struct {
|
||||
URL string `json:"url"`
|
||||
|
||||
@@ -200,17 +200,14 @@ func newModuleObservations(now time.Time) []conditions.Observation {
|
||||
// watchNewModules wakes the take-in of new modules for each module the builder or the indexer writes, until ctx ends, and
|
||||
// looks at every module once now. A bucket not there yet is said: the look at start and a first merge's word still work.
|
||||
func watchNewModules(ctx context.Context, open *stores) {
|
||||
inBackground(ctx, func() {
|
||||
inBackground(func() {
|
||||
takeNewModules(ctx, open)
|
||||
})
|
||||
// And, beside it, the start's pass of release 6 and the take-in of each registered module's new done key (novox/hq
|
||||
// ADR 0539 clauses 5, 7, 10 and 13; composed_start.go, composed_keys.go): the plans' buckets deleted, every node
|
||||
// composed, the replaced controller composed, and from now on each new fingerprint registered, composed and granted.
|
||||
inBackground(ctx, func() { composeAtStart(ctx, open) })
|
||||
inBackground(func() { composeAtStart(ctx, open) })
|
||||
watchDoneKeys(ctx, open)
|
||||
// And each assignment that comes to stand issues its node's bus accounts and the module's grant (ADR 0543): a module
|
||||
// built long before is assigned to one more node with no build done again (issue 1945, assignment_accounts.go).
|
||||
watchAssignmentAccounts(ctx, open)
|
||||
watchHavesForJudgement(ctx, open)
|
||||
// And the keep set keeps the builds every want names (ADR 0539 clause 1), with no plan left to name them.
|
||||
open.inventory.SetWanted(wantedOnTheBus)
|
||||
@@ -226,16 +223,16 @@ func watchNewModules(ctx context.Context, open *stores) {
|
||||
fmt.Printf("new modules: the builder's keys are not watched: %v\n", err)
|
||||
return
|
||||
}
|
||||
goServing(ctx, func() {
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
closeKeys()
|
||||
})
|
||||
}()
|
||||
}
|
||||
|
||||
// wakeNewModule takes module in when the catalogue does not hold it: never on the watch's own goroutine, which must keep
|
||||
// draining.
|
||||
func wakeNewModule(ctx context.Context, open *stores, module string) {
|
||||
inBackground(ctx, func() {
|
||||
inBackground(func() {
|
||||
entries, err := open.inventory.Catalogued(ctx)
|
||||
if err != nil {
|
||||
return
|
||||
@@ -253,17 +250,14 @@ func wakeNewModule(ctx context.Context, open *stores, module string) {
|
||||
// ends only once the work its mesh started has.
|
||||
var background sync.WaitGroup
|
||||
|
||||
// inBackground runs work beside the caller, counted in background and, under a serving controller, in its serving group,
|
||||
// so serve's stop waits for it and a serve that is stopping starts none (serving_group.go). A variable so a test of what a
|
||||
// walk does around a take-in may run it in the caller's turn and need not wait for it.
|
||||
var inBackground = func(ctx context.Context, work func()) {
|
||||
// inBackground runs work beside the caller, counted in background. A variable so a test of what a walk does around a
|
||||
// take-in may run it in the caller's turn and need not wait for it.
|
||||
var inBackground = func(work func()) {
|
||||
background.Add(1)
|
||||
if !goServing(ctx, func() {
|
||||
go func() {
|
||||
defer background.Done()
|
||||
work()
|
||||
}) {
|
||||
background.Done()
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// newModulesOf are the modules the indexer holds in a repository: what a first merge's word wakes.
|
||||
|
||||
@@ -59,17 +59,6 @@ func dataWords(o conditions.Observation) (what, module string) {
|
||||
}
|
||||
|
||||
var plainWordings = map[string]func(conditions.Observation) words{
|
||||
// A module's new build whose compositions or grant the controller could not write (composed_keys.go).
|
||||
kindNotComposed: worded(func(o conditions.Observation) words {
|
||||
m := "a module"
|
||||
if p := idPart(o, 0); o.Scope == conditions.ScopeBuild && p != "" {
|
||||
m = p // the module: raised in the build's scope, its id the module's name
|
||||
}
|
||||
return words{Headline: "A new build of " + m + " cannot be delivered",
|
||||
Explanation: fmt.Sprintf("A new build of %s was made, and the mesh could not prepare it for the machines, so "+
|
||||
"none can be given it yet. It is tried again every half minute; the details say what fails.", m),
|
||||
Resolved: "the new build of " + m + " can be delivered"}
|
||||
}),
|
||||
// Machines and their node-engine.
|
||||
"silent": worded(func(o conditions.Observation) words {
|
||||
m := machineOr(o, "a machine")
|
||||
|
||||
@@ -72,8 +72,6 @@ func issueMissingProcessAccounts(ctx context.Context, inv *inventory.Inventory,
|
||||
if len(m.ProcessBuses()) == 0 || mayIssue(m) != nil {
|
||||
return "", nil
|
||||
}
|
||||
// Only the missing: never a second password over one another caller minted between this look and the mint.
|
||||
ctx = onlyIfAbsent(ctx)
|
||||
minted, err := inv.BusMinted(ctx)
|
||||
if err != nil {
|
||||
return "", err
|
||||
|
||||
@@ -1,9 +1,12 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
)
|
||||
|
||||
// mailSenderWithProcess is mesh-mail's real manifest shape (novox/hq ADR 0436 §7, §8, the lead's fixture):
|
||||
@@ -22,6 +25,61 @@ func mailSenderWithProcess() catalogue.Manifest {
|
||||
return m
|
||||
}
|
||||
|
||||
// review round 1, finding 1 (1181): a module declaring no `broker` of its own at all — mesh-mail's real
|
||||
// shape, tools served by the tool runner — still passes mayIssue on its process's bus declaration alone.
|
||||
func TestAModuleWithNoBrokerAndOnlyAProcessMayBeIssued(t *testing.T) {
|
||||
if err := mayIssue(mailSenderWithProcess()); err != nil {
|
||||
t.Fatalf("a module with no broker of its own, whose process declares bus, was refused: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// 1181/AC1: a process with a bus declaration and no own secret broker-<process id> cannot be issued — the
|
||||
// same refusal mayIssue already gives a module with no `broker` (novox/hq ADR 0436 §2).
|
||||
func TestAProcessWithNoBrokerSecretCannotBeIssued(t *testing.T) {
|
||||
m := mailSenderWithProcess()
|
||||
delete(m.OwnSecrets, "broker-send")
|
||||
err := mayIssue(m)
|
||||
if err == nil {
|
||||
t.Fatal("a process with no broker-<id> own secret was issued an account")
|
||||
}
|
||||
for _, want := range []string{"mesh-mail", "send", "broker-send", "own-secrets"} {
|
||||
if !strings.Contains(err.Error(), want) {
|
||||
t.Errorf("the refusal does not say %q: %v", want, err)
|
||||
}
|
||||
}
|
||||
if err := mayIssue(mailSenderWithProcess()); err != nil {
|
||||
t.Errorf("a process declaring its own broker-<id> secret was refused: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// 1181/AC1: the account held for a process is the provisioner client's form — {url, fingerprint, node,
|
||||
// module, user, password} — named for the process, never the module's own account.
|
||||
func TestAProcesssHeldAccountIsTheProvisionersForm(t *testing.T) {
|
||||
user := broker.Principal{Kind: broker.KindProcess, Node: "novox", Module: "mesh-mail", Process: "send"}.Username()
|
||||
if user != "novox.mesh-mail.send" {
|
||||
t.Fatalf("the process's username: %q", user)
|
||||
}
|
||||
held, err := heldProcessAccount(broker.Broker{Fingerprint: "abc"}, "novox.example:4222", "novox", "mesh-mail",
|
||||
user, "a-password")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var got map[string]any
|
||||
if err := json.Unmarshal(held, &got); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
want := map[string]any{"url": "nats://novox.example:4222", "fingerprint": "abc", "node": "novox",
|
||||
"module": "mesh-mail", "user": "novox.mesh-mail.send", "password": "a-password"}
|
||||
for k, v := range want {
|
||||
if got[k] != v {
|
||||
t.Errorf("%s: %v, want %v", k, got[k], v)
|
||||
}
|
||||
}
|
||||
if len(got) != len(want) {
|
||||
t.Errorf("the held account carries %v; want exactly %v (never claims — a process holds no seat)", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
// 1181/AC1: composed, the process's own secret belongs to its own account at 0600 (novox/hq ADR 0436 §1, §2;
|
||||
// the owner mechanism is issue 1170's, proven generically there — this is mesh-mail's own shape).
|
||||
func TestAProcessesOwnSecretIsOwnedByItsOwnAccount(t *testing.T) {
|
||||
|
||||
+16
-39
@@ -59,9 +59,6 @@ func serve(ctx context.Context) (err error) {
|
||||
}
|
||||
defer open.Close()
|
||||
inv := open.inventory
|
||||
// Lent to every reader in this process that holds only the inventory, so none opens pools of its own per call
|
||||
// (issues 1945 and 1277: about 120 take-ins at once filled novox's store at 16:08 on 2026-10-25).
|
||||
defer shareStores(open)()
|
||||
// **What this build reads of the store's schema, on record** (novox/hq issue 352): the highest migration
|
||||
// it carries, by its version, so a gate that would put this build back later knows it reads the store
|
||||
// as it is then. A build that does not know its version records nothing, and is never put back.
|
||||
@@ -78,11 +75,11 @@ func serve(ctx context.Context) (err error) {
|
||||
"recorded, and a gate will never put it back\n", RunningBuildVar)
|
||||
}
|
||||
|
||||
// The controller's own identity store, the one its readers share (closed with open), not a second pool.
|
||||
ident, err := open.Identity(ctx)
|
||||
ident, err := openIdentity(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer ident.Close()
|
||||
|
||||
// Established at start rather than on first use. A control plane that cannot sign is one
|
||||
// whose declarations every node correctly refuses, and that should be a startup failure
|
||||
@@ -125,15 +122,7 @@ func serve(ctx context.Context) (err error) {
|
||||
defer theLease.release()
|
||||
ctx, stopActing := context.WithCancel(ctx)
|
||||
defer stopActing()
|
||||
// Every goroutine started from here is counted, and ended and waited for before serve returns (serving_group.go):
|
||||
// on the way out by error here, and at the end before anything else is undone.
|
||||
ctx, running := withServingGroup(ctx)
|
||||
stopRunning := func() {
|
||||
stopActing()
|
||||
running.stop()
|
||||
}
|
||||
defer stopRunning()
|
||||
goServing(ctx, func() {
|
||||
go func() {
|
||||
select {
|
||||
case <-lost:
|
||||
stopActing()
|
||||
@@ -143,7 +132,7 @@ func serve(ctx context.Context) (err error) {
|
||||
// asks the controller after this one rather than have a command started and killed with this
|
||||
// process (novox/hq issue 289).
|
||||
handingOver.Store(true)
|
||||
})
|
||||
}()
|
||||
defer func() {
|
||||
select {
|
||||
case <-lost:
|
||||
@@ -170,17 +159,6 @@ func serve(ctx context.Context) (err error) {
|
||||
if err := attachTheChain(ctx, inv, server); err != nil {
|
||||
return err
|
||||
}
|
||||
bus, isNATS := server.Bus().(link.OverNATS)
|
||||
if !isNATS {
|
||||
return errors.New("the mesh's verbs are served over the bus, and this control plane is not on it")
|
||||
}
|
||||
// The condition store, opened before anything that raises or reads a condition starts beside the serving, so none
|
||||
// of them meets conditionsFrom being written (issue 1945); closed after they all ended.
|
||||
keeper := openServingConditions(ctx, open, bus)
|
||||
if keeper != nil {
|
||||
defer closeConditions(keeper)
|
||||
defer stopRunning() // on an error return, the goroutines end before the keeper closes
|
||||
}
|
||||
|
||||
// The bus's own objects, asserted on every start. **Not created once at genesis**: a stream
|
||||
// somebody deleted, a mesh raised from a restored backup, or a bus whose data directory was
|
||||
@@ -195,7 +173,7 @@ func serve(ctx context.Context) (err error) {
|
||||
// And the pending assignments settled on a tick of their own (novox/hq ADR 0261): made once their module
|
||||
// is registered, ended with why when its build will not register it, raised and cleared as conditions.
|
||||
// Never by a read.
|
||||
goServing(ctx, func() { settlingPending(ctx, open) })
|
||||
go settlingPending(ctx, open)
|
||||
// The durations the core's bounds are set from are kept a month (novox/hq to-be 45 Phase 0).
|
||||
// And what the catalogue decided a build meant. The builder's own result is already handled
|
||||
// above; this is the other half — the control plane is the only one of the three that knows
|
||||
@@ -243,17 +221,19 @@ func serve(ctx context.Context) (err error) {
|
||||
"Those answer the reason when called; everything else is served as usual\n",
|
||||
catalogue.ControllerSeatName, strings.Join(behind, ", "))
|
||||
}
|
||||
bus, isNATS := server.Bus().(link.OverNATS)
|
||||
if !isNATS {
|
||||
return errors.New("the mesh's verbs are served over the bus, and this control plane is not on it")
|
||||
}
|
||||
// The hand-act log is counted for `status` on this connection rather than a new one a minute; no longer serving, the
|
||||
// connection is closed, and a verb run after in this process dials its own (a served test's leftover failed the
|
||||
// next test's `conditions` with "connection closed", issue 1945).
|
||||
handActConn = bus.Conn
|
||||
defer func() { handActConn = nil }()
|
||||
defer stopRunning() // on an error return, the goroutines end before handActConn is put back
|
||||
// And everything else this controller does on the bus for a moment (novox/hq issue 327).
|
||||
servingBus.Store(server.JetStream())
|
||||
// No longer serving: nothing is lent, and dead-letters says it is not read here (novox/hq issue 330).
|
||||
defer servingBus.Store(nil)
|
||||
defer stopRunning() // on an error return, the goroutines end before the bus is taken back
|
||||
// And says when it replaced a value given by hand (novox/hq ADR 0228).
|
||||
givenEvents = bus
|
||||
// And a pull request's merge check, asked when the forge announces its head and said when judged
|
||||
@@ -264,18 +244,18 @@ func serve(ctx context.Context) (err error) {
|
||||
return err
|
||||
}
|
||||
// Composed now and kept current, before the verb that answers from it is served.
|
||||
goServing(ctx, func() { statusFrom.keep(ctx) })
|
||||
go statusFrom.keep(ctx)
|
||||
// The facts snapshot a merge check is fed (novox/hq to-be 45 §9): kept current while this
|
||||
// controller holds the lease, read by the build seat from the artifact store.
|
||||
goServing(ctx, func() { exportingFacts(ctx, open, bus.Conn.ConnectedServerVersion) })
|
||||
go exportingFacts(ctx, open, bus.Conn.ConnectedServerVersion)
|
||||
// The record mirrors (novox/hq ADR 0289, design 57 F3): the assignments and each registered module's
|
||||
// declared config on the bus, written from the store only when a record changed, by the lease holder.
|
||||
// The provisioner's plan of each handed module, as its machine was last sent it (design 57 P6), in the module's
|
||||
// assignment.
|
||||
inv.SetProvisioningPlans(inv.SentProvisioningPlans)
|
||||
goServing(ctx, func() { mirroringRecords(ctx, open, inv, bus.Conn, bus) })
|
||||
go mirroringRecords(ctx, open, inv, bus.Conn, bus)
|
||||
// And the bucket of the files that travel by digest, judged against what each machine's latest declaration names (ADR 0486 §3).
|
||||
goServing(ctx, func() { sweepingDeclarationFiles(ctx, inv, bus.Conn) })
|
||||
go sweepingDeclarationFiles(ctx, inv, bus.Conn)
|
||||
// And the modules the catalogue does not hold yet, registered from the builder's first done key of each, woken by
|
||||
// the builder's and the indexer's writes (novox/hq ADR 0501 §6, issue 1851): nothing asks a first build.
|
||||
watchNewModules(ctx, open)
|
||||
@@ -307,7 +287,7 @@ func serve(ctx context.Context) (err error) {
|
||||
// signals table, what the bus says about itself, and the self-check. A store that cannot be opened
|
||||
// is said and the controller serves on: status then says the conditions cannot be read, and is
|
||||
// not well — louder than not serving at all, and the push that repairs the bus still runs.
|
||||
if stopWatching := watchTheMesh(ctx, open, server, bus, keeper); stopWatching != nil {
|
||||
if stopWatching := watchTheMesh(ctx, open, server, bus); stopWatching != nil {
|
||||
defer stopWatching()
|
||||
}
|
||||
stopServing, err := bus.ServeSeatTools(catalogue.ControllerSeatName, handlers, log.New(os.Stdout, "", log.LstdFlags))
|
||||
@@ -322,7 +302,7 @@ func serve(ctx context.Context) (err error) {
|
||||
}
|
||||
defer stopCLI()
|
||||
// Its read calls write no line each; how many each caller made is said once an hour (novox/hq issue 762).
|
||||
goServing(ctx, func() { sayReadCalls(ctx) })
|
||||
go sayReadCalls(ctx)
|
||||
// And says so on the bus (novox/hq ADR 0197): what it serves, as the NATS services protocol asks.
|
||||
stopAnnouncing, err := bus.Announce(seatAnnouncement(handlers), log.New(os.Stdout, "", log.LstdFlags))
|
||||
if err != nil {
|
||||
@@ -330,10 +310,7 @@ func serve(ctx context.Context) (err error) {
|
||||
}
|
||||
defer stopAnnouncing()
|
||||
|
||||
err = server.Serve(ctx)
|
||||
// Ended and waited for before the verbs stop being served and the globals are put back, not after.
|
||||
stopRunning()
|
||||
return err
|
||||
return server.Serve(ctx)
|
||||
}
|
||||
|
||||
// OverlayCIDRVar is the range the mesh allocates node addresses from.
|
||||
|
||||
@@ -2,15 +2,12 @@ package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sort"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/compose"
|
||||
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
@@ -27,8 +24,7 @@ import (
|
||||
// so no one call site sees every change; what every change has in common is that it is committed to the
|
||||
// store. So the serving controller reads the store every mirrorEvery and writes only what differs from what
|
||||
// the bucket holds (ADR 0289 §1): a quiet mesh writes nothing, and a change rolled back is never read. A
|
||||
// mirror is therefore at most mirrorEvery behind its store. The declared-packages mirror is the exception: it composes
|
||||
// every node, so it passes on a change of what it composes from, not on this tick (keepingDeclaredPackages, issue 1945).
|
||||
// mirror is therefore at most mirrorEvery behind its store.
|
||||
|
||||
// mirrorEvery is how often the mirrors are compared with the store: two small reads of the store.
|
||||
const mirrorEvery = 10 * time.Second
|
||||
@@ -49,31 +45,21 @@ func mirroringRecords(ctx context.Context, open *stores, inv *inventory.Inventor
|
||||
return e
|
||||
}
|
||||
sources := map[string]func(context.Context) (map[string][]byte, error){
|
||||
recordmirror.DeclaredConfig.Name: inv.DeclaredConfigMirror,
|
||||
recordmirror.Predefined.Name: inv.PredefinedMirror,
|
||||
recordmirror.DeclaredConfig.Name: inv.DeclaredConfigMirror,
|
||||
recordmirror.Predefined.Name: inv.PredefinedMirror,
|
||||
recordmirror.DeclaredPackages.Name: declaredPackagesMirror(open),
|
||||
}
|
||||
packages := keepingDeclaredPackages(ctx, open, api, announce, epoch, holds)
|
||||
held := false
|
||||
writers := map[string]*recordmirror.Writer{}
|
||||
failing := map[string]string{}
|
||||
tick := time.NewTicker(mirrorEvery)
|
||||
defer tick.Stop()
|
||||
events := &assignmentEvents{}
|
||||
for {
|
||||
now := holds()
|
||||
if now && !held {
|
||||
// The lease taken: the declared-packages mirror is brought to the stores at once, not at its next change.
|
||||
packages.poke()
|
||||
}
|
||||
held = now
|
||||
if now {
|
||||
if holds() {
|
||||
// The assignments are keys of the bus, written by the verbs that change them (novox/hq ADR 0408): this tick
|
||||
// keeps the provisioner's plan in their values and says what changed. They are no mirror of the store.
|
||||
keepingAssignments(ctx, inv, announce, epoch, events, failing)
|
||||
for _, m := range recordmirror.Mirrors() {
|
||||
if sources[m.Name] == nil {
|
||||
continue // the declared packages, kept on a change (keepingDeclaredPackages)
|
||||
}
|
||||
why := mirrorOnce(ctx, api, m, writers, sources[m.Name], announce, epoch, holds)
|
||||
if why != failing[m.Name] {
|
||||
if why != "" {
|
||||
@@ -193,263 +179,35 @@ func keepingAssignments(ctx context.Context, inv *inventory.Inventory, announce
|
||||
events.known = now
|
||||
}
|
||||
|
||||
// declaredPackagesMirror is the source of the declared-packages mirror (ADR 0507 §5) over a fresh reading of the stores.
|
||||
// declaredPackagesMirror is the source of the declared-packages mirror (ADR 0507 §5): each node's composition, asked as a
|
||||
// question that makes nothing, and the packages it declares with the modules that declare them. A node whose own set does
|
||||
// not compose has no key, and any other failure to read is the pass's error, so a key is never purged for a store that
|
||||
// could not be read.
|
||||
func declaredPackagesMirror(open *stores) func(context.Context) (map[string][]byte, error) {
|
||||
return func(ctx context.Context) (map[string][]byte, error) {
|
||||
in, err := inputFor(ctx, open)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return declaredPackagesFrom(ctx, in)
|
||||
}
|
||||
}
|
||||
|
||||
// declaredPackagesFrom is the declared-packages mirror over an Input: each node's composition, asked as a question that makes
|
||||
// nothing, and the packages it declares with the modules that declare them. A node whose own set does not compose has no key,
|
||||
// and any other failure to read is the pass's error, so a key is never purged for a store that could not be read.
|
||||
func declaredPackagesFrom(_ context.Context, in *meshInput) (map[string][]byte, error) {
|
||||
// Which module owns each resource is carried only for a node-engine that reads it, so every node is asked as one that does.
|
||||
asked := in.Input
|
||||
asked.Nodes = append([]compose.Node(nil), in.Nodes...)
|
||||
for i := range asked.Nodes {
|
||||
asked.Nodes[i].ReadsModules = true
|
||||
}
|
||||
out := map[string][]byte{}
|
||||
for _, n := range in.sortedNodes() {
|
||||
d, _, err := compose.ComposeIn(asked, n.Name, compose.Foreseeing)
|
||||
if compose.IsUnresolvable(err) {
|
||||
continue
|
||||
// Which module owns each resource is carried only for a node-engine that reads it, so every node is asked as one that does.
|
||||
asked := in.Input
|
||||
asked.Nodes = append([]compose.Node(nil), in.Nodes...)
|
||||
for i := range asked.Nodes {
|
||||
asked.Nodes[i].ReadsModules = true
|
||||
}
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the packages %s declares cannot be read: %w", n.Name, err)
|
||||
}
|
||||
if out[n.Name], err = inventory.DeclaredPackagesValue(n.Name, d.Resources, d.ModuleOf); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// **The declared-packages mirror passes on a change, not on a tick** (issue 1945 note 22). Its source composes every node in
|
||||
// full, about 45 s on the control node; on the 10 s tick it never rested and held the controller at 130-175 % CPU. Every
|
||||
// change that can move a node's packages is a change of its composition, which the controller writes into the bucket of the
|
||||
// compositions' digests, and of the assignments, which the verbs write into theirs: a watch of those two buckets wakes the
|
||||
// keeper. A pass reads the stores' Input once and composes only when the Input's digest differs from the one it last
|
||||
// mirrored, so a wake that changed nothing it composes from costs one reading. One pass at a time; the wakes during a pass
|
||||
// are one pass after it, at least packagesGap after the last began. A rare safety pass (packagesSafetyEvery) covers a
|
||||
// change no watched key says, and with an unchanged digest composes nothing either.
|
||||
const (
|
||||
packagesSafetyEvery = 10 * time.Minute
|
||||
packagesGap = 30 * time.Second
|
||||
)
|
||||
|
||||
// packagesTiming is when the declared-packages mirror passes: the safety pass's period and the least time between the
|
||||
// starts of two passes.
|
||||
type packagesTiming struct{ safety, gap time.Duration }
|
||||
|
||||
// packagesKeeper keeps the declared-packages mirror: read is the Input, compose the mirror's values from it, sync writes
|
||||
// those that differ from the bucket, holds says this controller holds the lease. Each a seam a test stands in for.
|
||||
type packagesKeeper struct {
|
||||
read func(ctx context.Context) (*meshInput, error)
|
||||
compose func(ctx context.Context, in *meshInput) (map[string][]byte, error)
|
||||
sync func(ctx context.Context, want map[string][]byte) error
|
||||
holds func() bool
|
||||
wake chan struct{}
|
||||
// last is the digest of the Input last mirrored whole; empty after a failure or while the lease is not held, so the
|
||||
// next pass composes.
|
||||
last string
|
||||
}
|
||||
|
||||
func newPackagesKeeper(read func(context.Context) (*meshInput, error),
|
||||
compose func(context.Context, *meshInput) (map[string][]byte, error),
|
||||
sync func(context.Context, map[string][]byte) error, holds func() bool) *packagesKeeper {
|
||||
return &packagesKeeper{read: read, compose: compose, sync: sync, holds: holds, wake: make(chan struct{}, 1)}
|
||||
}
|
||||
|
||||
// poke says something the mirror composes from may have changed; never blocks, and pokes before a pass are one.
|
||||
func (k *packagesKeeper) poke() {
|
||||
select {
|
||||
case k.wake <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
// pass brings the mirror to the Input, composing only when the Input differs from the one last mirrored.
|
||||
func (k *packagesKeeper) pass(ctx context.Context) error {
|
||||
if !k.holds() {
|
||||
k.last = ""
|
||||
return nil
|
||||
}
|
||||
in, err := k.read(ctx)
|
||||
if err != nil {
|
||||
k.last = ""
|
||||
return fmt.Errorf("the store cannot be read: %w", err)
|
||||
}
|
||||
digest := inputDigest(in)
|
||||
if digest != "" && digest == k.last {
|
||||
return nil
|
||||
}
|
||||
want, err := k.compose(ctx, in)
|
||||
if err != nil {
|
||||
k.last = ""
|
||||
return err
|
||||
}
|
||||
if err := k.sync(ctx, want); err != nil {
|
||||
k.last = ""
|
||||
return err
|
||||
}
|
||||
k.last = digest
|
||||
return nil
|
||||
}
|
||||
|
||||
// run passes once at its start, then on each wake and each safety period, until ctx ends. said hears why a pass failed,
|
||||
// and an empty line when one succeeds again.
|
||||
func (k *packagesKeeper) run(ctx context.Context, at packagesTiming, said func(why string)) {
|
||||
safety := time.NewTicker(at.safety)
|
||||
defer safety.Stop()
|
||||
failing := ""
|
||||
var began time.Time
|
||||
for {
|
||||
// The wakes so far are this pass's: drained before it reads, so a change after is a pass of its own.
|
||||
select {
|
||||
case <-k.wake:
|
||||
default:
|
||||
}
|
||||
began = time.Now()
|
||||
why := ""
|
||||
if err := k.pass(ctx); err != nil {
|
||||
why = err.Error()
|
||||
}
|
||||
if why != failing {
|
||||
said(why)
|
||||
failing = why
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-k.wake:
|
||||
case <-safety.C:
|
||||
}
|
||||
if rest := at.gap - time.Since(began); rest > 0 {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-time.After(rest):
|
||||
out := map[string][]byte{}
|
||||
for _, n := range in.sortedNodes() {
|
||||
d, _, err := compose.ComposeIn(asked, n.Name, compose.Foreseeing)
|
||||
if compose.IsUnresolvable(err) {
|
||||
continue
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// inputDigest is a digest of everything the composition reads (the Input, and why a node or the bus is refused): the same
|
||||
// for the same Input. Empty when it cannot be made, and then the pass composes.
|
||||
func inputDigest(in *meshInput) string {
|
||||
refused := map[string]string{}
|
||||
for node, err := range in.refused {
|
||||
if err != nil {
|
||||
refused[node] = err.Error()
|
||||
}
|
||||
}
|
||||
bus := ""
|
||||
if in.busRefused != nil {
|
||||
bus = in.busRefused.Error()
|
||||
}
|
||||
body, err := json.Marshal(struct {
|
||||
Input compose.Input
|
||||
Refused map[string]string
|
||||
BusRefused string
|
||||
}{in.Input, refused, bus})
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
sum := sha256.Sum256(body)
|
||||
return hex.EncodeToString(sum[:])
|
||||
}
|
||||
|
||||
// keepingDeclaredPackages keeps the declared-packages mirror for as long as ctx: woken by the watch of the compositions'
|
||||
// digests and of the assignments. It answers the keeper, whose poke a caller may also wake it with.
|
||||
func keepingDeclaredPackages(ctx context.Context, open *stores, api jetstream.JetStream, announce recordmirror.Announcer,
|
||||
epoch func() uint64, holds func() bool) *packagesKeeper {
|
||||
m := recordmirror.DeclaredPackages
|
||||
var w *recordmirror.Writer
|
||||
read := func(ctx context.Context) (*meshInput, error) {
|
||||
reading, cancel := context.WithTimeout(ctx, 2*time.Minute)
|
||||
defer cancel()
|
||||
return readComposeInput(reading, open)
|
||||
}
|
||||
sync := func(ctx context.Context, want map[string][]byte) error {
|
||||
writing, cancel := context.WithTimeout(ctx, 30*time.Second)
|
||||
defer cancel()
|
||||
if w == nil {
|
||||
kv, err := api.KeyValue(writing, m.Bucket)
|
||||
if err != nil {
|
||||
return fmt.Errorf("the bucket %s cannot be read: %v", m.Bucket, err)
|
||||
return nil, fmt.Errorf("the packages %s declares cannot be read: %w", n.Name, err)
|
||||
}
|
||||
w = &recordmirror.Writer{Mirror: m, KV: kv, Announce: announce, Epoch: epoch, Holds: holds}
|
||||
}
|
||||
lost := w.LostEvents
|
||||
r, err := w.Sync(writing, want)
|
||||
if r.Written+r.Removed > 0 {
|
||||
fmt.Printf("the %s mirror moved: %d written, %d removed\n", m.Name, r.Written, r.Removed)
|
||||
}
|
||||
if w.LostEvents > lost {
|
||||
fmt.Printf("the %s mirror lost %d event(s): the bucket holds the change, its notice was refused twice\n",
|
||||
m.Name, w.LostEvents-lost)
|
||||
}
|
||||
return err
|
||||
}
|
||||
k := newPackagesKeeper(read, declaredPackagesFrom, sync, holds)
|
||||
for _, bucket := range []string{sdkdelivery.ControllerComposedBucket, broker.AssignmentsBucket} {
|
||||
goServing(ctx, func() { wakeOnBucket(ctx, api, bucket, k.poke) })
|
||||
}
|
||||
goServing(ctx, func() {
|
||||
k.run(ctx, packagesTiming{safety: packagesSafetyEvery, gap: packagesGap}, func(why string) {
|
||||
if why != "" {
|
||||
fmt.Printf("the %s mirror cannot be kept: %s\n", m.Name, why)
|
||||
} else {
|
||||
fmt.Printf("the %s mirror is kept again\n", m.Name)
|
||||
}
|
||||
})
|
||||
})
|
||||
return k
|
||||
}
|
||||
|
||||
// wakeOnBucket calls wake on every write or delete of a key in the bucket, for as long as ctx: only the keys' metadata is
|
||||
// watched. A bucket not there yet, or a watch that ends, is watched again a minute later; the safety pass covers the gap.
|
||||
func wakeOnBucket(ctx context.Context, api jetstream.JetStream, bucket string, wake func()) {
|
||||
for {
|
||||
if err := watchBucketOnce(ctx, api, bucket, wake); err != nil && ctx.Err() == nil {
|
||||
fmt.Printf("the declared-packages mirror does not hear %s: %v\n", bucket, err)
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-time.After(time.Minute):
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// watchBucketOnce watches the bucket until its watch ends or ctx does.
|
||||
func watchBucketOnce(ctx context.Context, api jetstream.JetStream, bucket string, wake func()) error {
|
||||
kv, err := api.KeyValue(ctx, bucket)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
watch, err := kv.WatchAll(ctx, jetstream.UpdatesOnly(), jetstream.MetaOnly())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() { _ = watch.Stop() }()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil
|
||||
case e, ok := <-watch.Updates():
|
||||
if !ok {
|
||||
return fmt.Errorf("the watch ended")
|
||||
}
|
||||
if e != nil {
|
||||
wake()
|
||||
if out[n.Name], err = inventory.DeclaredPackagesValue(n.Name, d.Resources, d.ModuleOf); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,191 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/compose"
|
||||
)
|
||||
|
||||
// The declared-packages mirror composes only when what it composes from changed (issue 1945 note 22: it composed every node
|
||||
// in full every 10 s, a pass of about 45 s, so the controller on the control node never rested at 130-175 % CPU).
|
||||
|
||||
// scaled is a cadence of the mirror with a second taken as a millisecond, so a test's minute is 60 ms.
|
||||
func scaled() packagesTiming {
|
||||
return packagesTiming{safety: packagesSafetyEvery / 1000, gap: packagesGap / 1000}
|
||||
}
|
||||
|
||||
const scaledMinute = 60 * time.Millisecond
|
||||
|
||||
// fakeMesh is the keeper's input as a test changes it: each node's assigned modules, and the compositions counted by node.
|
||||
type fakeMesh struct {
|
||||
mu sync.Mutex
|
||||
assigned map[string][]string
|
||||
composed atomic.Int64 // node compositions
|
||||
synced atomic.Int64
|
||||
lastSync map[string][]byte
|
||||
// block, when set, holds the next composition until it is closed; entered is closed when that composition began.
|
||||
block chan struct{}
|
||||
entered chan struct{}
|
||||
}
|
||||
|
||||
func newFakeMesh(nodes int) *fakeMesh {
|
||||
f := &fakeMesh{assigned: map[string][]string{}}
|
||||
for i := 0; i < nodes; i++ {
|
||||
f.assigned[fmt.Sprintf("node%d", i)] = []string{"base"}
|
||||
}
|
||||
return f
|
||||
}
|
||||
|
||||
func (f *fakeMesh) assign(node, module string) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.assigned[node] = append(append([]string(nil), f.assigned[node]...), module)
|
||||
}
|
||||
|
||||
func (f *fakeMesh) keeper() *packagesKeeper {
|
||||
read := func(context.Context) (*meshInput, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
in := compose.Input{Catalogue: map[string]catalogue.Manifest{"base": {Module: "base", Version: "1"}}}
|
||||
for _, n := range nodesOfFake(f.assigned) {
|
||||
in.Nodes = append(in.Nodes, compose.Node{Name: n, Assigned: append([]string(nil), f.assigned[n]...)})
|
||||
}
|
||||
return &meshInput{Input: in}, nil
|
||||
}
|
||||
composeAll := func(_ context.Context, in *meshInput) (map[string][]byte, error) {
|
||||
f.mu.Lock()
|
||||
block, entered := f.block, f.entered
|
||||
f.block = nil
|
||||
f.mu.Unlock()
|
||||
if block != nil {
|
||||
close(entered)
|
||||
<-block
|
||||
}
|
||||
out := map[string][]byte{}
|
||||
for _, n := range in.Nodes {
|
||||
f.composed.Add(1)
|
||||
out[n.Name] = []byte(fmt.Sprint(n.Assigned))
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
syncAll := func(_ context.Context, want map[string][]byte) error {
|
||||
f.mu.Lock()
|
||||
f.lastSync = want
|
||||
f.mu.Unlock()
|
||||
f.synced.Add(1)
|
||||
return nil
|
||||
}
|
||||
return newPackagesKeeper(read, composeAll, syncAll, func() bool { return true })
|
||||
}
|
||||
|
||||
func nodesOfFake(m map[string][]string) []string {
|
||||
out := make([]string, 0, len(m))
|
||||
for k := range m {
|
||||
out = append(out, k)
|
||||
}
|
||||
for i := range out {
|
||||
for j := i + 1; j < len(out); j++ {
|
||||
if out[j] < out[i] {
|
||||
out[i], out[j] = out[j], out[i]
|
||||
}
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func runKeeper(t *testing.T, k *packagesKeeper) context.CancelFunc {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
defer close(done)
|
||||
k.run(ctx, scaled(), func(string) {})
|
||||
}()
|
||||
t.Cleanup(func() {
|
||||
cancel()
|
||||
<-done
|
||||
})
|
||||
return cancel
|
||||
}
|
||||
|
||||
func eventually(t *testing.T, what string, cond func() bool) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(time.Second)
|
||||
for !cond() {
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatalf("not within a second: %s", what)
|
||||
}
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
}
|
||||
|
||||
// 1945/AC-mirror-1: given a mesh whose input does not change, when a minute passes after the mirror's first pass, then it
|
||||
// composes no node (before: a pass every 10 s, each composing every node). The count per minute is said.
|
||||
func TestTheDeclaredPackagesMirrorComposesNothingWithoutAChange(t *testing.T) {
|
||||
const nodes = 5
|
||||
f := newFakeMesh(nodes)
|
||||
runKeeper(t, f.keeper())
|
||||
eventually(t, "the first pass", func() bool { return f.synced.Load() >= 1 })
|
||||
start := f.composed.Load()
|
||||
time.Sleep(scaledMinute)
|
||||
perMinute := f.composed.Load() - start
|
||||
t.Logf("node compositions in a minute of an unchanged mesh of %d nodes: %d", nodes, perMinute)
|
||||
if perMinute != 0 {
|
||||
t.Errorf("an unchanged mesh of %d nodes was composed %d times in a minute; want 0", nodes, perMinute)
|
||||
}
|
||||
}
|
||||
|
||||
// 1945/AC-mirror-2: given a pass that is running, when fifty changes arrive during it, then exactly one more pass follows,
|
||||
// never two at once and never one per change.
|
||||
func TestTheDeclaredPackagesMirrorPassesOncePerBurst(t *testing.T) {
|
||||
f := newFakeMesh(1)
|
||||
release := make(chan struct{})
|
||||
f.block, f.entered = release, make(chan struct{})
|
||||
k := f.keeper()
|
||||
runKeeper(t, k)
|
||||
<-f.entered
|
||||
for i := 0; i < 50; i++ {
|
||||
f.assign("node0", fmt.Sprintf("m%d", i))
|
||||
k.poke()
|
||||
}
|
||||
close(release)
|
||||
eventually(t, "the pass after the burst", func() bool { return f.synced.Load() >= 2 })
|
||||
time.Sleep(2*scaled().gap + 20*time.Millisecond)
|
||||
if got := f.synced.Load(); got != 2 {
|
||||
t.Errorf("a burst of 50 changes during a pass gave %d passes in all; want 2 (the running one and one after)", got)
|
||||
}
|
||||
}
|
||||
|
||||
// 1945/AC-mirror-3: given the mirror at rest, when an input changes and its event arrives, then the change is mirrored.
|
||||
func TestTheDeclaredPackagesMirrorMirrorsAChange(t *testing.T) {
|
||||
f := newFakeMesh(2)
|
||||
k := f.keeper()
|
||||
runKeeper(t, k)
|
||||
eventually(t, "the first pass", func() bool { return f.synced.Load() >= 1 })
|
||||
f.assign("node1", "viewer")
|
||||
k.poke()
|
||||
eventually(t, "the change mirrored", func() bool {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return string(f.lastSync["node1"]) == "[base viewer]"
|
||||
})
|
||||
}
|
||||
|
||||
// 1945/AC-mirror-4: the digest of the input is the same for the same input and differs when an assignment differs.
|
||||
func TestTheInputDigestFollowsTheInput(t *testing.T) {
|
||||
of := func(assigned ...string) *meshInput {
|
||||
return &meshInput{Input: compose.Input{Nodes: []compose.Node{{Name: "anchor", Assigned: assigned}}}}
|
||||
}
|
||||
a, b, c := inputDigest(of("base")), inputDigest(of("base")), inputDigest(of("base", "viewer"))
|
||||
if a == "" || a != b {
|
||||
t.Errorf("the same input gave the digests %q and %q", a, b)
|
||||
}
|
||||
if a == c {
|
||||
t.Errorf("an added assignment left the digest at %q", a)
|
||||
}
|
||||
}
|
||||
@@ -1,60 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// **Every goroutine the serving controller starts ends before serve returns** (issue 1945). A loop or a watch started
|
||||
// beside the serving runs until serve's context ends, and serve's stop waits for it before it closes the stores, the
|
||||
// bus and the condition keeper those loops read: a loop that outlived its serve met the next serve's globals, its
|
||||
// closed stores and its bus, and in the test process that is the next test (a race on conditionsFrom, a package that
|
||||
// passed every test and still exited failing). Where it is checked: the tests that run serve whole, under -race.
|
||||
|
||||
// servingGroup counts the goroutines one serve started. Once its stop began it starts nothing more: what would start is
|
||||
// work for a controller that is stopping.
|
||||
type servingGroup struct {
|
||||
mu sync.Mutex
|
||||
stopped bool
|
||||
wg sync.WaitGroup
|
||||
}
|
||||
|
||||
type servingGroupKey struct{}
|
||||
|
||||
// withServingGroup is ctx carrying a new group, which goServing under it counts.
|
||||
func withServingGroup(ctx context.Context) (context.Context, *servingGroup) {
|
||||
g := &servingGroup{}
|
||||
return context.WithValue(ctx, servingGroupKey{}, g), g
|
||||
}
|
||||
|
||||
// goServing runs work beside the caller, counted in the group ctx carries, so serve's stop waits for it. Under a group
|
||||
// whose stop began it runs nothing; under a context with no group (a part run alone, by a test or a verb) it is a plain
|
||||
// goroutine, as before. It says whether it started work.
|
||||
func goServing(ctx context.Context, work func()) bool {
|
||||
g, _ := ctx.Value(servingGroupKey{}).(*servingGroup)
|
||||
if g == nil {
|
||||
go work()
|
||||
return true
|
||||
}
|
||||
g.mu.Lock()
|
||||
if g.stopped {
|
||||
g.mu.Unlock()
|
||||
return false
|
||||
}
|
||||
g.wg.Add(1)
|
||||
g.mu.Unlock()
|
||||
go func() {
|
||||
defer g.wg.Done()
|
||||
work()
|
||||
}()
|
||||
return true
|
||||
}
|
||||
|
||||
// stop starts nothing more and waits for what was started; the caller ends their context first. A second stop waits
|
||||
// for nothing.
|
||||
func (g *servingGroup) stop() {
|
||||
g.mu.Lock()
|
||||
g.stopped = true
|
||||
g.mu.Unlock()
|
||||
g.wg.Wait()
|
||||
}
|
||||
@@ -1,86 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"github.com/novox/mesh-controller/internal/identity"
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
"github.com/novox/mesh-controller/internal/licences"
|
||||
)
|
||||
|
||||
// The serving controller's reads of the rest of the mesh open no pool per call: the licences and identity stores are the
|
||||
// controller's own, opened once and shared. On 2026-10-25 at 16:08 each take-in opened two pools of its own
|
||||
// (restOfTheMeshOf), about 120 at once, and novox's store refused every new connection (issues 1945 and 1277).
|
||||
func TestRestOfTheMeshOfOpensNoPoolPerCall(t *testing.T) {
|
||||
wasIdent, wasLic, wasRest := identityOpener, licencesOpener, restOfTheMeshOn
|
||||
t.Cleanup(func() { identityOpener, licencesOpener, restOfTheMeshOn = wasIdent, wasLic, wasRest })
|
||||
var identities, licencesOpened atomic.Int64
|
||||
identityOpener = func(context.Context) (*identity.Identity, error) {
|
||||
identities.Add(1)
|
||||
return &identity.Identity{}, nil
|
||||
}
|
||||
licencesOpener = func(context.Context) (*licences.Licences, error) {
|
||||
licencesOpened.Add(1)
|
||||
return &licences.Licences{}, nil
|
||||
}
|
||||
// The read of the Input reaches both stores, as readComposeInput does.
|
||||
restOfTheMeshOn = func(ctx context.Context, open *stores, _ map[string]catalogue.Manifest, _ string) (catalogue.World, error) {
|
||||
if _, err := open.Licences(ctx); err != nil {
|
||||
return catalogue.World{}, err
|
||||
}
|
||||
if _, err := open.Identity(ctx); err != nil {
|
||||
return catalogue.World{}, err
|
||||
}
|
||||
return catalogue.World{}, nil
|
||||
}
|
||||
inv := &inventory.Inventory{}
|
||||
serving := &stores{inventory: inv}
|
||||
unshare := shareStores(serving)
|
||||
t.Cleanup(unshare)
|
||||
|
||||
var wg sync.WaitGroup
|
||||
for range 120 {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
if _, err := restOfTheMeshOf(context.Background(), inv, nil, ""); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
if n := licencesOpened.Load(); n != 1 {
|
||||
t.Fatalf("opened the licences store %d times for 120 reads, want once", n)
|
||||
}
|
||||
if n := identities.Load(); n != 1 {
|
||||
t.Fatalf("opened the identity store %d times for 120 reads, want once", n)
|
||||
}
|
||||
if serving.licences == nil || serving.identity == nil {
|
||||
t.Fatal("the reads did not use the serving controller's stores")
|
||||
}
|
||||
}
|
||||
|
||||
// A process that shares no stores (a verb run as its own short process) still opens them for the call and lets go.
|
||||
func TestRestOfTheMeshOfOpensItsOwnWhereNoneAreShared(t *testing.T) {
|
||||
wasIdent, wasLic, wasRest := identityOpener, licencesOpener, restOfTheMeshOn
|
||||
t.Cleanup(func() { identityOpener, licencesOpener, restOfTheMeshOn = wasIdent, wasLic, wasRest })
|
||||
var opened atomic.Int64
|
||||
licencesOpener = func(context.Context) (*licences.Licences, error) { opened.Add(1); return &licences.Licences{}, nil }
|
||||
restOfTheMeshOn = func(ctx context.Context, open *stores, _ map[string]catalogue.Manifest, _ string) (catalogue.World, error) {
|
||||
_, err := open.Licences(ctx)
|
||||
return catalogue.World{}, err
|
||||
}
|
||||
inv := &inventory.Inventory{}
|
||||
for range 2 {
|
||||
if _, err := restOfTheMeshOf(context.Background(), inv, nil, ""); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if n := opened.Load(); n != 2 {
|
||||
t.Fatalf("opened %d times, want once per call", n)
|
||||
}
|
||||
}
|
||||
@@ -67,16 +67,16 @@ func keepAnnouncements(ctx context.Context, js *broker.JetStream, open *stores)
|
||||
return
|
||||
}
|
||||
if changed {
|
||||
inBackground(ctx, func() { composeAnnouncedReads(context.WithoutCancel(ctx), open) })
|
||||
inBackground(func() { composeAnnouncedReads(context.WithoutCancel(ctx), open) })
|
||||
}
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("following the states announced: %w", err)
|
||||
}
|
||||
goServing(ctx, func() {
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
consuming.Stop()
|
||||
})
|
||||
}()
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
@@ -210,7 +210,7 @@ func (l nudgingListener) Heard(ctx context.Context, report link.Report) (bool, e
|
||||
// And, once the move of the secrets into the vault has started, what is left of this machine's (design 57's V3,
|
||||
// issue 676): a machine asleep when it started is moved when it is next heard from. Off the report's path.
|
||||
if err == nil && l.open != nil && report.Superseded == "" && report.Rekey == nil && report.Node != "" {
|
||||
goServing(ctx, func() { moveWhenHeard(context.WithoutCancel(ctx), l.open, moveHeardOn(), report.Node) })
|
||||
go moveWhenHeard(context.WithoutCancel(ctx), l.open, moveHeardOn(), report.Node)
|
||||
}
|
||||
return news, err
|
||||
}
|
||||
|
||||
@@ -4,7 +4,6 @@ import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
@@ -164,13 +163,6 @@ func openLicences(ctx context.Context) (*licences.Licences, error) {
|
||||
return held, nil
|
||||
}
|
||||
|
||||
// identityOpener and licencesOpener open a store the first time a set of stores needs it; variables so a test counts
|
||||
// the pools opened.
|
||||
var (
|
||||
identityOpener = openIdentity
|
||||
licencesOpener = openLicences
|
||||
)
|
||||
|
||||
// stores is what one command has open.
|
||||
//
|
||||
// **Opened once, not once per machine.** Working out what a machine should be reaches the identity
|
||||
@@ -184,9 +176,6 @@ type stores struct {
|
||||
inventory *inventory.Inventory
|
||||
identity *identity.Identity
|
||||
licences *licences.Licences
|
||||
// mu keeps one opening of each store at a time: a serving controller's stores are shared by readers that run side
|
||||
// by side (sharedStores).
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
// open connects to the inventory, which every command that touches the mesh needs.
|
||||
@@ -200,12 +189,10 @@ func openStores(ctx context.Context) (*stores, error) {
|
||||
|
||||
// Identity is this control plane's own identity context, opened if it has not been.
|
||||
func (h *stores) Identity(ctx context.Context) (*identity.Identity, error) {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
if h.identity != nil {
|
||||
return h.identity, nil
|
||||
}
|
||||
opened, err := identityOpener(ctx)
|
||||
opened, err := openIdentity(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -215,12 +202,10 @@ func (h *stores) Identity(ctx context.Context) (*identity.Identity, error) {
|
||||
|
||||
// Licences is the context holding model access, opened if it has not been.
|
||||
func (h *stores) Licences(ctx context.Context) (*licences.Licences, error) {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
if h.licences != nil {
|
||||
return h.licences, nil
|
||||
}
|
||||
opened, err := licencesOpener(ctx)
|
||||
opened, err := openLicences(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -241,22 +226,3 @@ func (h *stores) Close() {
|
||||
h.inventory.Close()
|
||||
}
|
||||
}
|
||||
|
||||
// sharedStores are the stores a serving controller lends, by its inventory, to every reader in the process that holds
|
||||
// only the inventory (restOfTheMeshOf): one pool of each store for the controller, never one per call (issues 1945 and
|
||||
// 1277, 2026-10-25 16:08: about 120 take-ins opened two pools each and filled novox's store).
|
||||
var sharedStores sync.Map // *inventory.Inventory -> *stores
|
||||
|
||||
// shareStores lends open by its inventory until the function returned is called.
|
||||
func shareStores(open *stores) func() {
|
||||
sharedStores.Store(open.inventory, open)
|
||||
return func() { sharedStores.CompareAndDelete(open.inventory, open) }
|
||||
}
|
||||
|
||||
// sharedStoresOf are the stores lent for inv, if any are.
|
||||
func sharedStoresOf(inv *inventory.Inventory) (*stores, bool) {
|
||||
if open, ok := sharedStores.Load(inv); ok {
|
||||
return open.(*stores), true
|
||||
}
|
||||
return nil, false
|
||||
}
|
||||
|
||||
@@ -512,9 +512,7 @@ func (w *watchdogs) lostConsumers(ctx context.Context, heard []link.Advisory) (m
|
||||
for _, a := range deleted {
|
||||
_, err := w.js.Context().ConsumerInfo(a.Stream, a.Consumer)
|
||||
switch {
|
||||
// A consumer whose stream is gone is gone with it: the retired declarations' stream took every node's
|
||||
// consumer, and its deletion said each as deleted (issue 1945).
|
||||
case errors.Is(err, nats.ErrConsumerNotFound) || errors.Is(err, nats.ErrStreamNotFound):
|
||||
case errors.Is(err, nats.ErrConsumerNotFound):
|
||||
out[a.Stream+"."+a.Consumer] = want[a.Stream+"."+a.Consumer]
|
||||
case err != nil:
|
||||
return nil, fmt.Errorf("whether %s exists cannot be read: %w", link.ConsumerInWords(a.Stream, a.Consumer), err)
|
||||
@@ -545,9 +543,10 @@ func later(a, b time.Time) time.Time {
|
||||
return b
|
||||
}
|
||||
|
||||
// openServingConditions opens the condition store for the serving controller and makes it conditionsFrom. Nil when the store
|
||||
// could not be opened, which is said: status then says the conditions cannot be read.
|
||||
func openServingConditions(ctx context.Context, open *stores, bus link.OverNATS) *conditions.Keeper {
|
||||
// watchTheMesh opens the condition store and starts the watchdogs, the bus's advisories and the
|
||||
// self-check, for the serving controller; the returned function stops them. Nil when the store could
|
||||
// not be opened, which is said.
|
||||
func watchTheMesh(ctx context.Context, open *stores, server *link.Server, bus link.OverNATS) func() {
|
||||
keeper, err := keeperOn(ctx, bus.Conn, storeLooks{open.inventory})
|
||||
if err != nil {
|
||||
fmt.Printf("the condition store could NOT be opened, so nothing that goes wrong is kept or said, and "+
|
||||
@@ -555,22 +554,6 @@ func openServingConditions(ctx context.Context, open *stores, bus link.OverNATS)
|
||||
return nil
|
||||
}
|
||||
conditionsFrom = keeper
|
||||
return keeper
|
||||
}
|
||||
|
||||
// closeConditions flushes what the keeper holds and closes it, once nothing that raises a condition runs.
|
||||
func closeConditions(keeper *conditions.Keeper) {
|
||||
flushing, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
keeper.Close(flushing)
|
||||
}
|
||||
|
||||
// watchTheMesh starts the watchdogs, the bus's advisories and the self-check over the condition store's keeper, for the
|
||||
// serving controller; the returned function stops them and waits for them to end. Nil when there is no keeper.
|
||||
func watchTheMesh(ctx context.Context, open *stores, server *link.Server, bus link.OverNATS, keeper *conditions.Keeper) func() {
|
||||
if keeper == nil {
|
||||
return nil
|
||||
}
|
||||
logf := func(format string, args ...any) { fmt.Printf(format+"\n", args...) }
|
||||
stopHearing, err := server.HearAdvisories(logf)
|
||||
if err != nil {
|
||||
@@ -583,26 +566,26 @@ func watchTheMesh(ctx context.Context, open *stores, server *link.Server, bus li
|
||||
d := &doctor{open: open, js: server.JetStream(), keeper: keeper, teller: bus, watchdogs: w, host: host}
|
||||
w.doctor = d
|
||||
doctorFrom = d
|
||||
// Its own group as well as serve's: its stop waits for what it started.
|
||||
watching, stop := context.WithCancel(ctx)
|
||||
watching, running := withServingGroup(watching)
|
||||
goServing(watching, func() { w.keep(watching) })
|
||||
goServing(watching, func() { d.keep(watching) })
|
||||
go w.keep(watching)
|
||||
go d.keep(watching)
|
||||
// And the healers (novox/hq to-be 45 §7): what is open that a registered healer answers, repaired
|
||||
// under the lease and the brake, every act said.
|
||||
healers := newHealing(open, keeper, bus, server.JetStream())
|
||||
goServing(watching, func() { healers.keep(watching) })
|
||||
go healers.keep(watching)
|
||||
// And the asker (novox/hq ADR 0259): what needs the operator and names its answers is asked of them,
|
||||
// and the answer chosen is performed on its warrant.
|
||||
startAsking(watching, open, server, bus.Conn, keeper)
|
||||
goServing(watching, func() { forgettingOldHeals(watching, open.inventory) })
|
||||
go forgettingOldHeals(watching, open.inventory)
|
||||
fmt.Printf("watching the mesh: %d signal(s) every %s, %d probe(s) every %s; what is wrong is kept in %s "+
|
||||
"and said as %s events\n", len(watchedRows()), watchEvery, len(runnableProbes()), doctorEvery,
|
||||
broker.ConditionsBucket, conditions.Seat)
|
||||
return func() {
|
||||
stop()
|
||||
running.stop()
|
||||
stopHearing()
|
||||
flushing, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
keeper.Close(flushing)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -165,8 +165,7 @@ func HandPathWrites(nodes []string) ([]string, error) {
|
||||
// DeclarationsStream is the stream of each node's last declaration, `mesh.node.<node>.declare` (issue 107). Release 6
|
||||
// sends none (ADR 0539 clauses 5 and 11): the stream is kept, with each node's consumer, only so a node away during the
|
||||
// switch finds its last declaration and with it the new node-engine; the controller deletes it once every node's have
|
||||
// names the new node-engine, and from then on neither asserts it nor any node's consumer on it; nor makes it again once
|
||||
// it is gone from the bus for any reason, since a stream made anew would hold no node's last declaration (issue 1945).
|
||||
// names the new node-engine, and from then on neither asserts it nor any node's consumer on it.
|
||||
const DeclarationsStream = "NODES"
|
||||
|
||||
// RaiseAfterDeclarations is Raise once the declarations' stream is retired: the mesh's streams and consumers but the
|
||||
|
||||
@@ -114,37 +114,6 @@ func (i *Inventory) writeBusUser(ctx context.Context, u BusUser, hash string) er
|
||||
return nil
|
||||
}
|
||||
|
||||
// MintBusPasswordIfAbsent mints u's password only when no row names u yet, in one statement: of two callers that both
|
||||
// found the account missing, exactly one wins, and the other is told it lost (won false, no password) and must seal
|
||||
// nothing, so the hash the bus holds is always the password the node is sealed (novox/hq ADR 0543, the review of
|
||||
// issue 1945: the assign act and the assignments' watch once each minted, the later hash beside the earlier seal).
|
||||
func (i *Inventory) MintBusPasswordIfAbsent(ctx context.Context, u BusUser) (string, bool, error) {
|
||||
if u.Username == "" || u.Kind == "" {
|
||||
return "", false, errors.New("a bus user needs a username and a kind")
|
||||
}
|
||||
raw := make([]byte, 32)
|
||||
if _, err := rand.Read(raw); err != nil {
|
||||
return "", false, fmt.Errorf("cannot generate a bus password: %w", err)
|
||||
}
|
||||
password := base64.RawURLEncoding.EncodeToString(raw)
|
||||
hash, err := hashBusPassword(password)
|
||||
if err != nil {
|
||||
return "", false, fmt.Errorf("cannot hash a bus password: %w", err)
|
||||
}
|
||||
tag, err := i.store.Pool().Exec(ctx,
|
||||
`insert into bus_user (username, kind, node, module, password_hash)
|
||||
values ($1, $2, $3, $4, $5)
|
||||
on conflict (username) do nothing`,
|
||||
u.Username, u.Kind, u.Node, u.Module, string(hash))
|
||||
if err != nil {
|
||||
return "", false, fmt.Errorf("cannot record the bus user %s: %w", u.Username, err)
|
||||
}
|
||||
if tag.RowsAffected() == 0 {
|
||||
return "", false, nil
|
||||
}
|
||||
return password, true, nil
|
||||
}
|
||||
|
||||
func (i *Inventory) MintBusPassword(ctx context.Context, u BusUser) (string, error) {
|
||||
if u.Username == "" || u.Kind == "" {
|
||||
return "", errors.New("a bus user needs a username and a kind")
|
||||
|
||||
@@ -33,7 +33,7 @@ func TestABusPasswordIsReturnedOnceAndOnlyItsHashIsKept(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestMintingAgainRotatesAndMintingIfAbsentHasOneWinner(t *testing.T) {
|
||||
func TestMintingAgainRotatesRatherThanAddsAUser(t *testing.T) {
|
||||
inv := ForTest(t)
|
||||
ctx := context.Background()
|
||||
|
||||
@@ -62,50 +62,6 @@ func TestMintingAgainRotatesAndMintingIfAbsentHasOneWinner(t *testing.T) {
|
||||
if bcrypt.CompareHashAndPassword([]byte(hash), []byte(first)) == nil {
|
||||
t.Fatal("the previous password still verifies, so a rotation revoked nothing")
|
||||
}
|
||||
|
||||
// **Minting only if absent never rotates, and of several at once exactly one wins** (ADR 0543, the review of issue
|
||||
// 1945): the assign act and the assignments' watch may both find an account missing; the winner's password is the
|
||||
// one the store's hash verifies, so the one it seals is the one the bus accepts, and a loser seals nothing.
|
||||
if _, won, err := inv.MintBusPasswordIfAbsent(ctx, aBusUser("shop")); err != nil || won {
|
||||
t.Fatalf("a mint-if-absent over a minted account won %v (%v): it would replace a sealed password", won, err)
|
||||
}
|
||||
const racers = 8
|
||||
type minted struct {
|
||||
password string
|
||||
won bool
|
||||
err error
|
||||
}
|
||||
results := make(chan minted, racers)
|
||||
for range racers {
|
||||
go func() {
|
||||
p, won, err := inv.MintBusPasswordIfAbsent(ctx, aBusUser("racer"))
|
||||
results <- minted{p, won, err}
|
||||
}()
|
||||
}
|
||||
var winner string
|
||||
wins := 0
|
||||
for range racers {
|
||||
r := <-results
|
||||
if r.err != nil {
|
||||
t.Fatal(r.err)
|
||||
}
|
||||
if r.won {
|
||||
wins++
|
||||
winner = r.password
|
||||
} else if r.password != "" {
|
||||
t.Fatal("a caller that lost was handed a password to seal")
|
||||
}
|
||||
}
|
||||
if wins != 1 {
|
||||
t.Fatalf("%d of %d concurrent mints won, want exactly one", wins, racers)
|
||||
}
|
||||
stored, _, err := inv.BusUserHash(ctx, "one.racer")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := bcrypt.CompareHashAndPassword([]byte(stored), []byte(winner)); err != nil {
|
||||
t.Fatal("the stored hash is not the winner's password: the node would be sealed a password the bus refuses")
|
||||
}
|
||||
}
|
||||
|
||||
func TestForgettingANodeTakesItsBusUsersWithIt(t *testing.T) {
|
||||
|
||||
@@ -361,13 +361,3 @@ func (i *Inventory) AssignmentValue(ctx context.Context, node, module string) (c
|
||||
k, found, err := c.Assignments.Get(ctx, assignmentKey(node, module))
|
||||
return k.Value, found, err
|
||||
}
|
||||
|
||||
// WatchAssignments is the assignments bucket's raw entries, as the chain's watch gives them: every current value, a Caught
|
||||
// mark, then each write (novox/hq ADR 0543: an assignment that comes to stand issues its node's bus accounts).
|
||||
func (i *Inventory) WatchAssignments(ctx context.Context) (<-chan chain.Entry, error) {
|
||||
c, err := i.theChain(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return c.Assignments.Watch(ctx)
|
||||
}
|
||||
|
||||
@@ -1,82 +0,0 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
)
|
||||
|
||||
// A full store (SQLSTATE 53300, "too many clients already") is not asked again every 250 ms: Ready backs off,
|
||||
// doubling its wait up to a bound, and says once that the store is full (issues 1945 and 1277, 2026-10-25 16:08: the
|
||||
// controller's retries kept novox's store full).
|
||||
func TestReadyBacksOffOnAFullStore(t *testing.T) {
|
||||
wasPing, wasWait, wasSay := pingStore, readyWait, readySay
|
||||
t.Cleanup(func() { pingStore, readyWait, readySay = wasPing, wasWait, wasSay })
|
||||
asked := 0
|
||||
pingStore = func(context.Context, *Store) error {
|
||||
asked++
|
||||
if asked <= 8 {
|
||||
return &pgconn.PgError{Code: "53300", Message: "sorry, too many clients already"}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
var waits []time.Duration
|
||||
readyWait = func(_ context.Context, d time.Duration) error { waits = append(waits, d); return nil }
|
||||
var said []string
|
||||
readySay = func(line string) { said = append(said, line) }
|
||||
|
||||
if err := (&Store{context: "example"}).Ready(context.Background(), time.Hour); err != nil {
|
||||
t.Fatalf("Ready: %v", err)
|
||||
}
|
||||
if len(waits) != 8 {
|
||||
t.Fatalf("waited %d times, want 8: %v", len(waits), waits)
|
||||
}
|
||||
for i := 1; i < len(waits); i++ {
|
||||
if waits[i] < waits[i-1] {
|
||||
t.Fatalf("wait %d (%s) is shorter than the one before (%s): %v", i, waits[i], waits[i-1], waits)
|
||||
}
|
||||
}
|
||||
if waits[1] <= waits[0] {
|
||||
t.Fatalf("Ready does not back off on a full store: %v", waits)
|
||||
}
|
||||
if most := waits[len(waits)-1]; most > fullStoreMostWait {
|
||||
t.Fatalf("the wait grew past its bound %s: %v", fullStoreMostWait, waits)
|
||||
}
|
||||
full := 0
|
||||
for _, l := range said {
|
||||
if strings.Contains(l, "full") {
|
||||
full++
|
||||
}
|
||||
}
|
||||
if full != 1 {
|
||||
t.Fatalf("said that the store is full %d times, want once: %q", full, said)
|
||||
}
|
||||
}
|
||||
|
||||
// A store that does not answer for another reason (not up yet) is still asked again every 250 ms.
|
||||
func TestReadyWaitsShortlyOnAStoreNotUp(t *testing.T) {
|
||||
wasPing, wasWait, wasSay := pingStore, readyWait, readySay
|
||||
t.Cleanup(func() { pingStore, readyWait, readySay = wasPing, wasWait, wasSay })
|
||||
asked := 0
|
||||
pingStore = func(context.Context, *Store) error {
|
||||
asked++
|
||||
if asked <= 3 {
|
||||
return context.DeadlineExceeded
|
||||
}
|
||||
return nil
|
||||
}
|
||||
var waits []time.Duration
|
||||
readyWait = func(_ context.Context, d time.Duration) error { waits = append(waits, d); return nil }
|
||||
readySay = func(string) {}
|
||||
if err := (&Store{context: "example"}).Ready(context.Background(), time.Hour); err != nil {
|
||||
t.Fatalf("Ready: %v", err)
|
||||
}
|
||||
for _, w := range waits {
|
||||
if w != 250*time.Millisecond {
|
||||
t.Fatalf("waits %v, want 250ms each", waits)
|
||||
}
|
||||
}
|
||||
}
|
||||
+6
-47
@@ -24,7 +24,6 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/envfile"
|
||||
@@ -214,7 +213,7 @@ func (s *Store) Pool() *pgxpool.Pool { return s.pool }
|
||||
|
||||
// Close releases the connections.
|
||||
func (s *Store) Close() {
|
||||
if s != nil && s.pool != nil {
|
||||
if s.pool != nil {
|
||||
s.pool.Close()
|
||||
}
|
||||
}
|
||||
@@ -228,10 +227,8 @@ func (s *Store) Close() {
|
||||
func (s *Store) Ready(ctx context.Context, within time.Duration) error {
|
||||
deadline := time.Now().Add(within)
|
||||
var last error
|
||||
wait := 250 * time.Millisecond
|
||||
saidFull := false
|
||||
for {
|
||||
err := pingStore(ctx, s)
|
||||
err := s.pool.Ping(ctx)
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
@@ -243,48 +240,10 @@ func (s *Store) Ready(ctx context.Context, within time.Duration) error {
|
||||
return fmt.Errorf(
|
||||
"the %s store did not answer within %s: %w", s.context, within, last)
|
||||
}
|
||||
// A full store (SQLSTATE 53300) is not asked again every 250 ms: each ask is one more connection it refuses
|
||||
// while it is full. The wait doubles up to fullStoreMostWait, and the journal says once why (issues 1945 and
|
||||
// 1277, 2026-10-25 16:08: the controller's own retries kept the store full).
|
||||
if isFull(err) {
|
||||
if !saidFull {
|
||||
readySay(fmt.Sprintf("the %s store is full (too many clients already); asked again at a growing "+
|
||||
"interval, at most every %s, for at most %s in all", s.context, fullStoreMostWait, within))
|
||||
saidFull = true
|
||||
}
|
||||
} else {
|
||||
wait = 250 * time.Millisecond
|
||||
}
|
||||
if err := readyWait(ctx, wait); err != nil {
|
||||
return errors.Join(err, last)
|
||||
}
|
||||
if isFull(err) {
|
||||
wait = min(wait*2, fullStoreMostWait)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return errors.Join(ctx.Err(), last)
|
||||
case <-time.After(250 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// pingStore asks the store once whether it answers; a variable so a test needs no database.
|
||||
var pingStore = func(ctx context.Context, s *Store) error { return s.pool.Ping(ctx) }
|
||||
|
||||
// readyWait waits d or until ctx ends; a variable so a test waits in no real time.
|
||||
var readyWait = func(ctx context.Context, d time.Duration) error {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-time.After(d):
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// readySay says a line of Ready's wait in the journal; a variable so a test reads it.
|
||||
// isFull says whether err is the store refusing a connection because it is full (SQLSTATE 53300).
|
||||
func isFull(err error) bool {
|
||||
var pg *pgconn.PgError
|
||||
return errors.As(err, &pg) && pg.Code == "53300"
|
||||
}
|
||||
|
||||
// fullStoreMostWait bounds the wait between two asks of a full store.
|
||||
const fullStoreMostWait = 5 * time.Second
|
||||
|
||||
var readySay = func(line string) { fmt.Fprintln(os.Stderr, line) }
|
||||
|
||||
Reference in New Issue
Block a user