Issue a node's bus accounts and grant when its assignment comes to stand (issue 1945, ADR 0543)
ADR 0539 clause 7 issues a module's credential and grant only when its build is done. A module built long before and assigned to one more node never has a build done again: on 2026-10-25 at 17:19 mesh-build-node and mesh-build-go were assigned to ace and composed for it, the bus's user list never named ace's accounts, and both runtimes restarted every 5 s (one with no credential, one refused by the bus). The assign act composes the assigned node alone, never the node holding the bus, and writes no grant. So a write of the assignments bucket is the event: when a key comes to stand (new, or ended or pending before), the lease holder issues the module's accounts on that node that are not minted (never re-issued), composes the node holding the bus and writes the grants, through the composer the done keys' take-ins use. A rewrite of a standing assignment does nothing; the start's look settles every standing one, one after another. A step of research 072's grants (memberships and credentials heard from placement).
This commit is contained in:
@@ -0,0 +1,299 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"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:
|
||||
// it issues what is missing and, when it issued anything, composes the bus's node and writes the grants.
|
||||
|
||||
// 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
|
||||
// issue issues the module's accounts on the node that are not minted; issued says it minted one, line what it did.
|
||||
issue func(ctx context.Context, 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, fresh bool) (string, error) {
|
||||
if !s.holds(ctx) || 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
|
||||
}
|
||||
issued, line, err := s.issue(ctx, node, module)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("its bus accounts on %s are not issued: %w", node, err)
|
||||
}
|
||||
if !issued && !fresh {
|
||||
return "", nil
|
||||
}
|
||||
holder, err := s.busHolder(ctx)
|
||||
if err != nil {
|
||||
return line, fmt.Errorf("the node holding the bus cannot be read: %w", err)
|
||||
}
|
||||
if holder != "" {
|
||||
if err := s.compose(ctx, []string{holder}); err != nil {
|
||||
return line, fmt.Errorf("the bus's user list on %s is not composed: %w", holder, err)
|
||||
}
|
||||
}
|
||||
if err := s.grants(ctx); err != nil {
|
||||
return line, err
|
||||
}
|
||||
said := module + " on " + node + ": "
|
||||
if line != "" {
|
||||
said += line + "; "
|
||||
}
|
||||
if holder != "" {
|
||||
said += "the bus's user list composed on " + holder + ", "
|
||||
}
|
||||
return said + "its grant 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, node, module string) (bool, string, error) {
|
||||
shelf, err := inv.Catalogue(ctx)
|
||||
if err != nil {
|
||||
return false, "", fmt.Errorf("the catalogue cannot be read: %w", err)
|
||||
}
|
||||
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 {
|
||||
epoch, err := theLease.epoch(ctx)
|
||||
return err == nil && epoch > 0
|
||||
},
|
||||
issue: func(ctx context.Context, node, module string) (bool, string, error) {
|
||||
return issueAssignmentAccounts(ctx, inv, 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
|
||||
|
||||
// settleAssignment settles one assignment until it works or ctx ends: a failure is said and tried again every
|
||||
// assignmentAccountsRetry, never dropped, since no later write of the key may come.
|
||||
func settleAssignment(ctx context.Context, steps assignmentAccountSteps, node, module string, fresh bool) {
|
||||
for {
|
||||
said, err := steps.settle(ctx, node, module, fresh)
|
||||
if err == nil {
|
||||
if said != "" {
|
||||
fmt.Printf("assignments: %s\n", said)
|
||||
}
|
||||
return
|
||||
}
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
fmt.Printf("assignments: %s on %s: %v; tried again in %s\n", module, node, 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)
|
||||
turns := &assignmentTurns{}
|
||||
goServing(ctx, func() {
|
||||
for {
|
||||
var atStart []chain.Assignment
|
||||
if err := watchAssignmentsOnce(ctx, open.inventory.WatchAssignments, turns, func(a chain.Assignment, fresh bool) {
|
||||
if !fresh {
|
||||
atStart = append(atStart, a)
|
||||
return
|
||||
}
|
||||
inBackground(ctx, func() { settleAssignment(ctx, steps, a.Node, a.Module, true) })
|
||||
}, func() {
|
||||
// The start's look, one assignment after another: a reading of the catalogue each, never all at once
|
||||
// (issues 1945 and 1277: a burst of take-ins once filled the store).
|
||||
look := atStart
|
||||
atStart = nil
|
||||
inBackground(ctx, func() {
|
||||
for _, a := range look {
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
settleAssignment(ctx, steps, a.Node, a.Module, false)
|
||||
}
|
||||
})
|
||||
}); err != nil && ctx.Err() == nil {
|
||||
fmt.Printf("assignments: the assignments are not heard, so a new assignment's bus accounts wait: %v\n", err)
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-time.After(time.Minute):
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// watchAssignmentsOnce watches the assignments until the watch or ctx ends, handing each assignment that comes to stand
|
||||
// to settle: fresh for one written after the current values were read, not fresh for one of them, and calling read once
|
||||
// the current values are all handed (the chain's Caught mark).
|
||||
func watchAssignmentsOnce(ctx context.Context, watch func(context.Context) (<-chan chain.Entry, error),
|
||||
turns *assignmentTurns, 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
|
||||
read()
|
||||
}
|
||||
continue
|
||||
}
|
||||
var value []byte
|
||||
if !e.Deleted {
|
||||
value = e.Value
|
||||
}
|
||||
if a, turned := turns.heard(e.Key, value); turned {
|
||||
settle(a, !current)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,251 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"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.
|
||||
func TestAnAssignmentTurnsWhenItComesToStand(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
|
||||
}
|
||||
cases := []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}}},
|
||||
}
|
||||
for _, c := range cases {
|
||||
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)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// 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; for an assignment that just came to stand even when its accounts were minted before (a module
|
||||
// assigned again keeps its account), and never by a controller that does not hold the lease.
|
||||
func TestSettlingAnAssignmentIssuesComposesTheBusAndGrants(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
module string
|
||||
holds, fresh bool
|
||||
issued bool
|
||||
issueErr, composeE error
|
||||
holder string
|
||||
wantComposed []string
|
||||
wantGrants bool
|
||||
wantErr string
|
||||
}{
|
||||
{name: "a new assignment whose account is not minted", module: "mesh-build-node", holds: true, fresh: true,
|
||||
issued: true, holder: "novox", wantComposed: []string{"novox"}, wantGrants: true},
|
||||
{name: "a new assignment whose account was minted before", module: "mesh-build-go", holds: true, fresh: true,
|
||||
holder: "novox", wantComposed: []string{"novox"}, wantGrants: true},
|
||||
{name: "the start's look issuing a missing account", module: "mesh-build-node", holds: true, issued: true,
|
||||
holder: "novox", wantComposed: []string{"novox"}, wantGrants: true},
|
||||
{name: "the start's look with every account minted does nothing", module: "mesh-build-go", holds: true,
|
||||
holder: "novox"},
|
||||
{name: "no lease, nothing", module: "mesh-build-node", fresh: true, issued: true, holder: "novox"},
|
||||
{name: "the controller's own module is the terminal's", module: controllerModule, holds: true, fresh: true,
|
||||
issued: true, holder: "novox"},
|
||||
{name: "no bus holder: the grants still", module: "mesh-build-node", holds: true, fresh: true, issued: true,
|
||||
wantGrants: true},
|
||||
{name: "an issue that fails is the error, nothing composed", module: "mesh-build-node", holds: true, fresh: true,
|
||||
issueErr: errors.New("the bus cannot be reached"), holder: "novox", wantErr: "the bus cannot be reached"},
|
||||
{name: "a composition that fails is the error, no grant", module: "mesh-build-node", holds: true, fresh: true,
|
||||
issued: true, holder: "novox", composeE: errors.New("store refused"), wantComposed: []string{"novox"},
|
||||
wantErr: "store refused"},
|
||||
}
|
||||
for _, c := range cases {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
var composed []string
|
||||
issuedFor, grants := "", false
|
||||
steps := assignmentAccountSteps{
|
||||
holds: func(context.Context) bool { return c.holds },
|
||||
issue: func(_ context.Context, node, module string) (bool, string, error) {
|
||||
issuedFor = module + "@" + node
|
||||
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 = true; return nil },
|
||||
}
|
||||
said, err := steps.settle(t.Context(), "ace", c.module, c.fresh)
|
||||
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 strings.Join(composed, ",") != strings.Join(c.wantComposed, ",") {
|
||||
t.Fatalf("composed %v, want %v", composed, c.wantComposed)
|
||||
}
|
||||
if grants != c.wantGrants {
|
||||
t.Fatalf("grants written %v, want %v", grants, c.wantGrants)
|
||||
}
|
||||
skipped := !c.holds || c.module == controllerModule
|
||||
if skipped != (issuedFor == "") {
|
||||
t.Fatalf("issue asked for %q (lease held %v, module %s)", issuedFor, c.holds, c.module)
|
||||
}
|
||||
if c.wantGrants && c.wantErr == "" && !strings.Contains(said, "its grant written") {
|
||||
t.Fatalf("said %q, want it to name the grant", said)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// The account 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.
|
||||
func TestAModulesOwnAccountOnANode(t *testing.T) {
|
||||
client := aProvisioner()
|
||||
delete(client.OwnSecrets, "broker")
|
||||
cases := []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},
|
||||
}
|
||||
for _, c := range cases {
|
||||
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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// The 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 rewrite of one that stands is handed nothing.
|
||||
func TestTheWatchHandsTheStartsLookThenEachNewAssignment(t *testing.T) {
|
||||
at := time.Date(2026, 10, 25, 16, 19, 42, 0, time.UTC)
|
||||
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})
|
||||
|
||||
type handed struct {
|
||||
key string
|
||||
fresh bool
|
||||
}
|
||||
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{},
|
||||
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{}
|
||||
}
|
||||
}
|
||||
if h := next(); h != (handed{"g14.mesh-build-go", false}) {
|
||||
t.Fatalf("the start's look handed %+v", h)
|
||||
}
|
||||
select {
|
||||
case <-read:
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("the current values were never said read")
|
||||
}
|
||||
replanned := chain.Assignment{Node: "g14", Module: "mesh-build-go", Assigned: at, Provisioning: json.RawMessage(`{"plan":2}`)}
|
||||
put(replanned)
|
||||
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 len(read) != 0 {
|
||||
t.Fatal("the current values were said read more than once")
|
||||
}
|
||||
}
|
||||
@@ -208,6 +208,9 @@ func watchNewModules(ctx context.Context, open *stores) {
|
||||
// composed, the replaced controller composed, and from now on each new fingerprint registered, composed and granted.
|
||||
inBackground(ctx, 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)
|
||||
|
||||
@@ -361,3 +361,13 @@ 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)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user