Compare commits

...
Author SHA1 Message Date
mesh-tools 0fbb5ef359 Merge pull request 'The controller gives its lease back at the stop signal and says a takeover once, so a controller delivery raises no urgent alarm (issue 906, ADR 0558)' (#478) from fix/906-lease-release into main 2026-10-27 00:07:46 +00:00
jschoubben 8f84731896 Name the epoch given back in the lease's delete marker, so only a holder's own release counts as a handover (issue 906, ADR 0558 §3, review round 1)
mesh/delivery merged: source.main moved to novox/mesh-controller@0fbb5ef35991
mesh/check being carried to 8f847318963c on 0fbb5ef35991 | from 2dafbe60e939
A plain delete marker on the lease key is also written by a command's own
lease and by a take given back over the floor; counted as a handover, it
would record a crashed holder's epoch released and say nothing. Release now
writes the marker itself, compare-and-set on the key's revision, with the
header Mesh-Lease-Released naming its epoch, and the next holder ends only
that epoch as released. A takeover whose predecessor cannot be read is now
raised as an expiry, not only printed.
2026-10-27 00:55:30 +01:00
jschoubben fbaa463a38 Give the controller lease back at the stop signal and say a takeover once, so a controller delivery raises no urgent alarm (issue 906, ADR 0558)
The 2026-10-26 replacements showed the lease was given back, but last: the
shutdown took 10 s, the gate had closed at half the key's age, and the release
recorded the epoch released only through the gate, so the next holder found
it expired and S12 said it, urgent, for an hour.

Now the stop signal gives the key back first and records the epoch released
whenever the delete succeeded; a candidate reads the key's delete marker on
the bus to tell a handover from an expiry; and a real takeover (lost or
expired) is raised and cleared once by the controller that took it, instead
of an hour of S12 observation.

Research 072 piece moved: the lease's takeover read from its own key on the
bus and said as an event (03-domains, mesh-controller row; 04-findings 2,
the watchdogs' age reading).
Quick fix weighed: drop the gate check from the release record only; it
leaves the release last in the shutdown and the hour-long alarm for a real
takeover.
2026-10-27 00:49:13 +01:00
mesh-tools 2dafbe60e9 Merge pull request 'Broker tests read a consumer's pending count through pendingOf wherever they expect a count after a publish (issue 1126)' (#477) from fix/1126-tests-wait-for-consumer-counts into main 2026-10-26 22:57:59 +00:00
jschoubben 8d58afea6c Broker tests read a consumer's pending count through pendingOf, which waits up to 10 s, wherever they expect a count after a publish (issue 1126)
mesh/delivery merged: source.main moved to novox/mesh-controller@2dafbe60e939
mesh/check being carried to 8d58afea6cb8 on 2dafbe60e939 | from b611f7375a47
The server counts a consumer's pending after it acknowledges the publish, so a read at once failed the try of
main on a loaded ace (TestAResetOfTheEventsConsumerLeavesTheMergesAndTheirConsumerAlone, 2026-10-26 22:24 UTC).
Fixed for the kind, not the one test: all six reads of a positive count go through the package's own pendingOf,
whose deadline was 1 s.

Research 'The controller, NATS-native': none; this changes tests only, not a part of the controller.
2026-10-26 23:28:03 +01:00
mesh-tools b611f7375a Merge pull request 'Pin mesh-sdk main, whose seats promise own and settle, so the catalogue passes the module check again (issue 1126)' (#476) from fix/1126-controller-pins-sdk-main into main 2026-10-26 22:08:19 +00:00
jschoubben 253bbcea2d Pin mesh-sdk main f178323, whose delivery seat promises settle beside the vault's own, so the catalogue check passes against mesh-catalog main f0f7b673 (issue 1126)
mesh/delivery merged: source.main moved to novox/mesh-controller@b611f7375a47
mesh/check checking 253bbcea2da8 on b611f7375a47
Moves no piece of research 'The controller, NATS-native': a repin of the seat catalogue, no controller code changed.
2026-10-26 23:03:47 +01:00
mesh-tools 2c6382346f Merge pull request 'Pin the mesh-sdk whose vault seat promises own, so the catalogue passes the module check again (issue 1126)' (#475) from fix/1126-vault-seat-own into main 2026-10-26 22:00:46 +00:00
jschoubben b8cafccaf6 Pin mesh-sdk at e78e881, whose vault seat promises own, so the catalogue check passes against mesh-catalog main (issue 1126)
mesh/delivery merged: source.main moved to novox/mesh-controller@2c6382346f48
mesh/delivery-group group fix/1126-vault-seat-own merged: every member is merged
mesh/check failed: unit: github.com/novox/mesh-controller/cmd/mesh-controller.TestModuleCheckPassesTheCatalogue: the catalogue does not pass its own c…
Moves no piece of research 'The controller, NATS-native': a repin for a seat verb list, no controller code changed.
2026-10-26 22:40:34 +01:00
mesh-tools 82d212a2a1 Merge pull request 'The controller answers each node's placement from its own store, the one verb of its exit (issue 1126)' (#474) from feat/1126-nodes-compose into main 2026-10-26 21:11:21 +00:00
jschoubben df8ebf6a77 Pin mesh-sdk 9796444 (l2-next: feat/1126-nodes-compose merged with sdk main #101, the standing-test race fixes) (issue 1126)
mesh/delivery merged: source.main moved to novox/mesh-controller@82d212a2a1ff
mesh/delivery-group group feat/1126-nodes-compose merged: every member is merged
mesh/check checking df8ebf6a77c1 on 82d212a2a1ff | novox/mesh-controller#474 in group feat/1126-nodes-compose
2026-10-26 22:05:41 +01:00
jschoubben dc414fbfef Pin mesh-sdk 5f45175 (l2-next: feat/1126-nodes-compose merged with sdk main) after merging main (issue 1126) 2026-10-26 21:59:21 +01:00
jschoubben fd891688ac Pin mesh-sdk feat/1126-nodes-compose 03dce12, the group's head: the checker refuses a pin that is neither trunk nor the group's head (issue 1126)
mesh/delivery closed: a newer head of the same pull request
mesh/delivery-group group feat/1126-nodes-compose checking: 4 of 6 member(s) ready
mesh/check checking fd891688acdf on 2ecfbb3a29cb | novox/mesh-controller#474 in group feat/1126-nodes-compose
2026-10-26 20:35:49 +01:00
jschoubben 9d8a5a7ede Review of L2, round 2: the placement export is checked within an existing integration test over the same fresh mesh, so it adds no test of its own (the test budget; issue 1126)
mesh/delivery closed: a newer head of the same pull request
mesh/check checking 9d8a5a7ede90 on 2ecfbb3a29cb | novox/mesh-controller#474 in group feat/1126-nodes-compose
2026-10-26 20:23:51 +01:00
jschoubben 4e9e5c27d4 Review of L2: the export's test reads each printed placement back as the file the seed writes, at its version (S1); pin mesh-sdk 3968849, CAPTURED taken again (issue 1126)
mesh/delivery closed: a newer head of the same pull request
mesh/check failed: unit: ..test: ✔ a transient empty list for four passes retires nothing; five stable passes retire, data kept (290.408814ms) | 4e9e5…
2026-10-26 20:12:49 +01:00
jschoubben 7c8a2e03dd Pin mesh-sdk feat/1126-nodes-compose 9713c30: placement carries a settings-form module's values (issue 1126)
mesh/delivery closed: a newer head of the same pull request
mesh/check failed: ace: novox/mesh-catalog#505 does not merge into the trunk: files in conflict: .gitignore | 7c8a2e03dd8e on 2ecfbb3a29cb | unit 1245…
2026-10-26 19:40:43 +01:00
jschoubben 89f52a6504 Pin mesh-sdk 53c2def, the group's head, and take the SDK's conformance copy again at it (issue 1126)
mesh/delivery closed: a newer head of the same pull request
mesh/check failed: unit: git.novox.be/novox/mesh-catalog/checks/nobinaries.TestTheCatalogueHoldsNoCompiledProgram: modules/mesh-config/cmd/mesh-config…
2026-10-26 19:28:39 +01:00
jschoubben fb2f5a2962 The controller answers each node's placement from its own store, read-only, for the seed of the catalogue's placement files: the one verb of its exit (issue 1126, ADR 0555 §6; the operator 2026-10-26 19:06:40) 2026-10-26 19:16:09 +01:00
mesh-tools 2ecfbb3a29 Merge pull request 'The controller takes mesh-sdk a7a1fbd3, so its composer gives build modules the node's CPU share (issue 1806, ADR 0553/0554 change B)' (#473) from fix/controller-sdk-cpu-share into main 2026-10-26 15:41:03 +00:00
jschoubben 9ad8715871 Bump mesh-sdk to a7a1fbd3 so the build seats' runtimes are composed to yield (issue 1806)
mesh/delivery merged: source.main moved to novox/mesh-controller@2ecfbb3a29cb
mesh/check pass 9ad8715871ea on 2ecfbb3a29cb | carried from e1a55f099b4e | unit 1023 run, 171 kept, 2 skipped | integration 1701 run, 214 kept, 9 skip…
The controller is the only composer, and its pin (10b6e202) predates mesh-sdk #95: without the bump the
build runtimes never get yields {priority idle, cpu-share} or MESH_BUILD_CPU_SHARE (ADR 0553, ADR 0554,
change B). The operator said yes to this one change of the frozen controller on 2026-10-26 15:54.

A composition at both pins over every node's assignment differs only in mesh-build-go, mesh-build-node and
mesh-build-image (the yields and the share word); every other composition is byte-identical. The SDK's
conformance capture moves with the pin, as testdata/beside/CAPTURED asks.

Research 072 (the controller, NATS-native): no piece moved; a pin bump the operator allowed.
2026-10-26 16:34:55 +01:00
mesh-tools e1a55f099b Merge pull request 'Seat users pulled by kind may read their own seat's dead letters (issue 734)' (#472) from fix/734-lost-ask into main 2026-10-26 13:40:43 +00:00
jschoubben ccc7ba334c broker: DeadLetterReads grants only what a pulled ordered consumer needs, as mesh-sdk buscompose (issue 734, review round 1)
mesh/delivery merged: source.main moved to novox/mesh-controller@e1a55f099b4e
mesh/delivery-group group fix/734-lost-ask merged: every member is merged
mesh/check checking ccc7ba334c82 on 08a261ac1a6d | novox/mesh-controller#472 in group fix/734-lost-ask
NATS-native research piece: none moved; the copy stays in step with mesh-sdk until mesh-grants composes the list.
2026-10-26 14:17:00 +01:00
jschoubben 65d5ed3367 broker: a user of a seat pulled by kind reads its seat's dead letters, as mesh-sdk buscompose composes it (issue 734)
The checker hears from DEAD_LETTERS which of its asks a build seat gave up on, and ends the try or asks the check again;
the controller composes today's user list, so its copy of the composer carries the grant too. NATS-native research piece:
none moved; the composition itself moves to mesh-grants (ADR 0547), and this keeps the copy in step with mesh-sdk.
2026-10-26 13:51:35 +01:00
mesh-tools 08a261ac1a Merge pull request 'Declarations test: the indexer stand-in answers the trunk value at the trunk's commit alone (issue 1945)' (#471) from fix/declarations-test-stand-in into main 2026-10-26 12:30:18 +00:00
jschoubben 23226384e5 Answer the indexer's trunk value at the trunk's commit alone in the declarations test, so its take-in records mesh-host's build (issue 1945)
mesh/delivery merged: source.main moved to novox/mesh-controller@08a261ac1a6d
mesh/check being carried to 23226384e529 on 08a261ac1a6d | from 350f8ebdfe5f
The stub answered the trunk for every commit, so since registering on the
fingerprint (666d1b68, #465) mesh-host, registered at an older commit, looked
already at the trunk's fingerprint: the take-in only moved its commit, no build
was recorded, and the fingerprints never named the new node-engine. The real
indexer holds no value at a commit it never indexed.

(cherry picked from commit 779533764b)
2026-10-26 13:29:04 +01:00
mesh-tools 350f8ebdfe Merge pull request 'Broker: the bus node may follow the grants part's sealed user list and the grants part may write it, the operator's second one-grant exception (issue 1945)' (#470) from feat/1945-user-list-switch into main 2026-10-26 12:21:44 +00:00
23 changed files with 506 additions and 67 deletions
+42 -6
View File
@@ -13,6 +13,7 @@ import (
"github.com/nats-io/nats.go/jetstream" "github.com/nats-io/nats.go/jetstream"
"github.com/novox/mesh-controller/internal/broker" "github.com/novox/mesh-controller/internal/broker"
"github.com/novox/mesh-controller/internal/conditions"
"github.com/novox/mesh-controller/internal/inventory" "github.com/novox/mesh-controller/internal/inventory"
"github.com/novox/mesh-controller/internal/lease" "github.com/novox/mesh-controller/internal/lease"
"github.com/novox/mesh-controller/internal/link" "github.com/novox/mesh-controller/internal/link"
@@ -72,6 +73,14 @@ type actor struct {
// lease over in a second rather than in a renewal of five (novox/hq issue 591). A renewal is given // lease over in a second rather than in a renewal of five (novox/hq issue 591). A renewal is given
// the lease's own RenewEvery to answer however often it is made, so a slow bus does not lose it. // the lease's own RenewEvery to answer however often it is made, so a slow bus does not lose it.
renewEvery, poll time.Duration renewEvery, poll time.Duration
// releasing is held for the whole of a release, so a second caller — the stop signal's and serve's own
// deferred one — waits until the lease is given back and its epoch recorded, never past it to the store's
// close (novox/hq issue 906).
releasing sync.Mutex
// keeper and inv are where a takeover is said (sayTakeover): set once the serving controller's condition
// store opens; nil before, and in a command.
keeper *conditions.Keeper
inv *inventory.Inventory
} }
// theLease is this process's standing to the lease. // theLease is this process's standing to the lease.
@@ -174,8 +183,11 @@ func (a *actor) forACommand(ctx context.Context) (uint64, error) {
return holder.Epoch, nil return holder.Epoch, nil
} }
// release gives back what this process holds, at its end. // release gives back what this process holds, at its end — or at the stop signal, before the rest of the
// process is taken down (serveUnderTheLease). A second call waits for the first to finish.
func (a *actor) release() { func (a *actor) release() {
a.releasing.Lock()
defer a.releasing.Unlock()
a.mu.Lock() a.mu.Lock()
closing := a.close closing := a.close
a.close = nil a.close = nil
@@ -266,20 +278,40 @@ func (a *actor) serveUnderTheLease(ctx context.Context, inv *inventory.Inventory
} }
a.mu.Lock() a.mu.Lock()
a.close = func() { a.close = func() {
if held, err := l.Epoch(); err == nil { // Given back first, then recorded: an epoch is recorded released only when its key is gone, and
// whatever the gate says by then — a shutdown longer than half the key's age closed it, and the epoch
// was left open for the next holder to find "expired" (novox/hq issue 906, 2026-10-26).
if given := l.Release(context.Background()); given != 0 {
ending, cancel := context.WithTimeout(context.Background(), 5*time.Second) ending, cancel := context.WithTimeout(context.Background(), 5*time.Second)
if err := inv.EndEpoch(ending, held, inventory.EpochReleased); err != nil { if err := inv.EndEpoch(ending, given, inventory.EpochReleased); err != nil {
say("how epoch %d ended could not be recorded: %v", held, err) say("how epoch %d ended could not be recorded: %v", given, err)
} }
cancel() cancel()
} }
l.Release(context.Background())
js.Close() js.Close()
} }
a.mu.Unlock() a.mu.Unlock()
// **The stop signal gives the lease back first** (novox/hq issue 906): the moment ctx ends, before the rest
// of the process is taken down, the gate closes and the key goes, so the next controller takes the lease
// at once and the epoch is recorded released. Waiting for the process's own end left the key unrenewed for
// as long as the shutdown took — ten seconds on 2026-10-26, past the gate's half-age.
go func() {
<-ctx.Done()
if a.holds(l) {
say("stopping: the controller lease is given back first")
}
a.release()
}()
return gone, nil return gone, nil
} }
// holds is whether this process holds l now.
func (a *actor) holds(l *lease.Lease) bool {
a.mu.Lock()
defer a.mu.Unlock()
return a.held == l
}
// took is the lease taken: recorded, earlier epochs nobody gave back ended as expired, kept. // took is the lease taken: recorded, earlier epochs nobody gave back ended as expired, kept.
func (a *actor) took(ctx context.Context, l *lease.Lease, inv *inventory.Inventory, epoch uint64, gone chan struct{}) { func (a *actor) took(ctx context.Context, l *lease.Lease, inv *inventory.Inventory, epoch uint64, gone chan struct{}) {
a.mu.Lock() a.mu.Lock()
@@ -288,15 +320,19 @@ func (a *actor) took(ctx context.Context, l *lease.Lease, inv *inventory.Invento
h := holderOf(instance) h := holderOf(instance)
recording, cancel := context.WithTimeout(ctx, 10*time.Second) recording, cancel := context.WithTimeout(ctx, 10*time.Second)
expired, err := inv.TookEpoch(recording, inventory.Epoch{Epoch: epoch, Instance: h.Instance, Host: h.Host, expired, err := inv.TookEpoch(recording, inventory.Epoch{Epoch: epoch, Instance: h.Instance, Host: h.Host,
Build: h.Build, Taken: time.Now()}) Build: h.Build, Taken: time.Now()}, l.GivenBack())
cancel() cancel()
if err != nil { if err != nil {
fmt.Printf("epoch %d could not be recorded as taken, so a stale refusal from it will not name it: %v\n", epoch, err) fmt.Printf("epoch %d could not be recorded as taken, so a stale refusal from it will not name it: %v\n", epoch, err)
} }
if given := l.GivenBack(); given != 0 {
fmt.Printf("the lease was given back by the controller of epoch %d: taken at once, at epoch %d\n", given, epoch)
}
for _, e := range expired { for _, e := range expired {
fmt.Printf("the controller of epoch %d (%s) stopped renewing the lease without giving it back: it is "+ fmt.Printf("the controller of epoch %d (%s) stopped renewing the lease without giving it back: it is "+
"taken over at epoch %d\n", e.Epoch, e.Instance, epoch) "taken over at epoch %d\n", e.Epoch, e.Instance, epoch)
} }
a.sayTakeover(ctx, epoch)
go l.Keep(ctx) go l.Keep(ctx)
go func() { go func() {
<-l.Lost() <-l.Lost()
@@ -78,6 +78,8 @@ func TestTheRegistryTrustAndEveryImageFollowThePortTheNodeGaveTheStore(t *testin
func TestTheControlPlaneIsToldWhereTheNodePutTheStoreAndTheBroker(t *testing.T) { func TestTheControlPlaneIsToldWhereTheNodePutTheStoreAndTheBroker(t *testing.T) {
open := aMesh(t) open := aMesh(t)
// Issue 1126: the placement export over this fresh mesh, before anything below changes it.
placementExportIsEachNodesPlacementFromTheStore(t, open)
ctx := t.Context() ctx := t.Context()
raw, err := os.ReadFile("../../module.json") raw, err := os.ReadFile("../../module.json")
if err != nil { if err != nil {
@@ -13,6 +13,7 @@ import (
"github.com/nats-io/nats.go/jetstream" "github.com/nats-io/nats.go/jetstream"
"github.com/novox/mesh-controller/internal/broker" "github.com/novox/mesh-controller/internal/broker"
"github.com/novox/mesh-controller/internal/builderkeys"
"github.com/novox/mesh-controller/internal/identity" "github.com/novox/mesh-controller/internal/identity"
"github.com/novox/mesh-controller/internal/link" "github.com/novox/mesh-controller/internal/link"
"github.com/novox/mesh-controller/internal/testbus" "github.com/novox/mesh-controller/internal/testbus"
@@ -76,13 +77,18 @@ func aComposingMesh(t *testing.T) *composingMesh {
return m return m
} }
// trunkKeys are a done key and the indexer's trunk value naming it. // trunkKeys are a done key and the indexer's trunk value naming it. The indexer holds that value at the trunk's commit
// alone, as the real one does: answering it for any commit made a module registered at an older commit look already
// registered at the trunk's fingerprint, so the take-in only moved its commit and recorded no build (issue 1945).
type trunkKeys struct { type trunkKeys struct {
keyBuckets keyBuckets
trunk sdkdelivery.Module trunk sdkdelivery.Module
} }
func (k trunkKeys) ModuleAt(context.Context, string, string) (sdkdelivery.Module, error) { func (k trunkKeys) ModuleAt(_ context.Context, _ string, commit string) (sdkdelivery.Module, error) {
if commit != "" && !sameCommitID(commit, k.trunk.Commit) {
return sdkdelivery.Module{}, builderkeys.ErrNoValue
}
return k.trunk, nil return k.trunk, nil
} }
+15 -2
View File
@@ -89,15 +89,28 @@ func TestTwoControllersOneActs(t *testing.T) {
t.Fatalf("B, waiting, may act: %v", err) t.Fatalf("B, waiting, may act: %v", err)
} }
// A hands over, as a controller being replaced does: B takes the lease at once, at a higher epoch. // A is stopped, as its service manager stops a controller being replaced (novox/hq issue 906): the stop
// alone gives the lease back, before anything else of A is taken down, and B takes it within a second, at
// a higher epoch — no wait for the key's age, and no condition said.
stopA() stopA()
a.release() stopped := time.Now()
var lostB <-chan struct{} var lostB <-chan struct{}
select { select {
case lostB = <-tookB: case lostB = <-tookB:
case <-time.After(10 * time.Second): case <-time.After(10 * time.Second):
t.Fatal("B did not take the lease A gave back") t.Fatal("B did not take the lease A gave back")
} }
if waited := time.Since(stopped); waited > time.Second {
t.Fatalf("B took the lease %s after A was stopped: the stop did not give it back first", waited)
}
a.release() // what serve's own end does after: waits for the release under way, gives nothing twice
conditionsOfB := conditions.NewInMemory()
keeperOfB := conditions.NewKeeper(ctx, conditions.Options{Store: conditionsOfB, History: conditionsOfB})
b.watchTakeovers(ctx, inv, keeperOfB)
keeperOfB.Close(ctx)
if said, err := conditionsOfB.Since(ctx, stopped.Add(-time.Minute)); err != nil || len(said) != 0 {
t.Fatalf("a lease given back is said as a condition: %+v, %v", said, err)
}
select { select {
case <-lostA: case <-lostA:
default: default:
+6 -1
View File
@@ -133,9 +133,14 @@ func nodeCommand(ctx context.Context, args []string) error {
// refuse (novox/hq issue 358). Read-only on the node and here, so a verb may run it too. // refuse (novox/hq issue 358). Read-only on the node and here, so a verb may run it too.
return nodeJudgeDirectories(ctx, open, args[1:]) return nodeJudgeDirectories(ctx, open, args[1:])
case "placement-export":
// Each node's placement as the catalogue's placement file says it (novox/hq ADR 0555 §6): read-only, so a verb
// may run it.
return placementExport(ctx, open, args[1:])
default: default:
return fmt.Errorf("node has no %q; it has add, list, show, public-domain, account, agent-account, hand-over, "+ return fmt.Errorf("node has no %q; it has add, list, show, public-domain, account, agent-account, hand-over, "+
"setuid-search and judge-directories", args[0]) "setuid-search, judge-directories and placement-export", args[0])
} }
} }
+50
View File
@@ -0,0 +1,50 @@
package main
import (
"context"
"errors"
"sort"
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// placementExport prints each node's placement as the catalogue's placement file says it (novox/hq ADR 0555 §6, issue
// 1126): what a person decided of the node and where its modules run, read by this controller's own code from its own
// store, as JSON keyed by node. The one verb the controller gains for its exit (the operator, 2026-10-26 19:06:40, ADR
// 0547 §1's one exception besides §5's grant): read once by the seed of the placement files, whose pull request merges
// only when its composition check composes every node equal to what it runs. It reads the same Input every composition
// of this controller reads (readMeshAndNodes, readSettings) and writes nothing.
//
// It moves a piece of the research "The controller, NATS-native": the controller's last node data (node rows, placement,
// ports, pins, bindings, config placement) leaves it for the catalogue, before its exit at ADR 0547's third milestone.
func placementExport(ctx context.Context, open *stores, args []string) error {
if len(args) > 1 {
return errors.New("node placement-export [<name>]")
}
in, _, err := readMeshAndNodes(ctx, open)
if err != nil {
return err
}
if _, err := readSettings(ctx, open, &in); err != nil {
return err
}
var names []string
if len(args) == 1 {
names = []string{args[0]}
} else {
for _, n := range in.Nodes {
names = append(names, n.Name)
}
sort.Strings(names)
}
out := make(map[string]delivery.Placement, len(names))
for _, name := range names {
p, err := frombus.Export(in, name)
if err != nil {
return err
}
out[name] = p
}
return printJSON(out)
}
@@ -0,0 +1,71 @@
//go:build integration
package main
import (
"encoding/json"
"strings"
"testing"
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/overlay"
)
// Issue 1126 (novox/hq ADR 0555 §6, the operator's one export verb of 2026-10-26 19:06:40): `node placement-export`
// answers, from the real store, each node's placement as the catalogue's placement file says it: what a person decided
// of the node (its endpoint, site, hub and address on the private network) and its modules in the order the composition
// reads them, exactly frombus.Export of the input this controller composes from; one node when named; a node the store
// lacks is refused. Checked within TestTheControlPlaneIsToldWhereTheNodePutTheStoreAndTheBroker, over the same fresh
// mesh, so the export adds no integration test of its own (the test budget).
func placementExportIsEachNodesPlacementFromTheStore(t *testing.T, open *stores) {
t.Helper()
ctx := t.Context()
in, _, err := readMeshAndNodes(ctx, open)
if err != nil {
t.Fatal(err)
}
if _, err := readSettings(ctx, open, &in); err != nil {
t.Fatal(err)
}
var all map[string]sdkdelivery.Placement
if err := json.Unmarshal([]byte(stdoutOf(t, func() error { return placementExport(ctx, open, nil) })), &all); err != nil {
t.Fatal(err)
}
if len(all) != 2 {
t.Fatalf("exported %d nodes, want anchor and laptop", len(all))
}
for i, name := range []string{"anchor", "laptop"} {
got := all[name]
d := got.Decided
if d.Endpoint != name+".example:51820" || d.Site != "here" || d.Hub != (name == "anchor") || d.Address != []string{"10.77.0.1", "10.77.0.2"}[i] {
t.Errorf("%s's decided facts are %+v", name, d)
}
if a := got.Assigned(); len(a) == 0 || !strings.Contains(strings.Join(a, ","), overlay.Name) {
t.Errorf("%s's modules are %v, want the private network among them", name, a)
}
// What it prints is the file the seed writes: at the placement file's version, read back by its reader.
if raw, _ := json.Marshal(got); got.V != sdkdelivery.Version {
t.Errorf("%s is exported at version %d: %s", name, got.V, raw)
} else if _, err := sdkdelivery.DecodePlacement(raw); err != nil {
t.Errorf("%s's export does not read as a placement file: %v", name, err)
}
want, err := frombus.Export(in, name)
if err != nil {
t.Fatal(err)
}
if !sdkdelivery.SamePlacement(got, want) {
t.Errorf("%s's export differs from frombus.Export of the controller's own input:\n%+v\n%+v", name, got, want)
}
}
var one map[string]sdkdelivery.Placement
if err := json.Unmarshal([]byte(stdoutOf(t, func() error { return placementExport(ctx, open, []string{"laptop"}) })), &one); err != nil {
t.Fatal(err)
}
if _, ok := one["laptop"]; !ok || len(one) != 1 {
t.Errorf("one node named exported %v", one)
}
if err := placementExport(ctx, open, []string{"stranger"}); err == nil {
t.Error("a node the store lacks was exported")
}
}
+3
View File
@@ -180,6 +180,9 @@ func serve(ctx context.Context) (err error) {
if keeper != nil { if keeper != nil {
defer closeConditions(keeper) defer closeConditions(keeper)
defer stopRunning() // on an error return, the goroutines end before the keeper closes defer stopRunning() // on an error return, the goroutines end before the keeper closes
// The takeover of the lease said as it happened, raised and cleared (novox/hq issue 906).
theLease.watchTakeovers(ctx, inv, keeper)
defer theLease.watchTakeovers(ctx, nil, nil) // the keeper closes after this: nothing said into it then
} }
// The bus's own objects, asserted on every start. **Not created once at genesis**: a stream // The bus's own objects, asserted on every start. **Not created once at genesis**: a stream
+8 -1
View File
@@ -349,6 +349,11 @@ func (a *verbArguments) commandLine() ([]string, error) {
return nil, err return nil, err
} }
return []string{"node", "judge-directories", str("node")}, nil return []string{"node", "judge-directories", str("node")}, nil
case "placement-export":
if node := str("node"); node != "" {
return []string{"node", "placement-export", node}, nil
}
return []string{"node", "placement-export"}, nil
case "modules": case "modules":
return []string{"module", "list", "--json"}, nil return []string{"module", "list", "--json"}, nil
case "seats": case "seats":
@@ -1400,7 +1405,9 @@ func seatAnnouncement(handlers map[string]link.ToolHandler) micro.Info {
// nodeReads are the `node` subcommands a verb may run: the ones that only read. // nodeReads are the `node` subcommands a verb may run: the ones that only read.
var nodeReads = map[string]bool{"list": true, "show": true, var nodeReads = map[string]bool{"list": true, "show": true,
// Read-only on the node too: the engine judges its directories and changes nothing (novox/hq issue 358). // Read-only on the node too: the engine judges its directories and changes nothing (novox/hq issue 358).
"judge-directories": true} "judge-directories": true,
// Each node's placement as the catalogue's placement file says it, read from the store (novox/hq ADR 0555 §6).
"placement-export": true}
// flagsOnly says a command line's rest names no subcommand: empty, or beginning with a flag. For a command // flagsOnly says a command line's rest names no subcommand: empty, or beginning with a flag. For a command
// with no subcommands every word is a flag, its value or a name it reads. // with no subcommands every word is a flag, its value or a name it reads.
+109 -11
View File
@@ -1,6 +1,7 @@
package main package main
import ( import (
"context"
"fmt" "fmt"
"sort" "sort"
"strings" "strings"
@@ -147,9 +148,10 @@ var signalsTable = []signalRow{
return newestOf(f.machines, func(m machineFacts) time.Time { return m.toolsHeard }) return newestOf(f.machines, func(m machineFacts) time.Time { return m.toolsHeard })
}}, }},
{Row: "S12", Signal: "the controller lease renewed", Emitter: "controller", Trigger: "every 5 s", {Row: "S12", Signal: "the controller lease renewed", Emitter: "controller", Trigger: "every 5 s",
Bound: "15 s (the key's age); a holder that lost the lease, or stopped renewing and was taken over, and a " + Bound: "15 s (the key's age); a holder that lost the lease or stopped renewing it is said once by the " +
"lease bucket found raised again from nothing, are said for an hour after; a controller serving without " + "controller that took it over, raised and cleared as it takes it, and for an hour after while none " +
"the lease, for as long as it does", "holds it; a lease given back is never said; a lease bucket found raised again from nothing is said " +
"for an hour after; a controller serving without the lease, for as long as it does",
Kind: "lease-lost", Severity: conditions.Urgent, Phase: 2, Kind: "lease-lost", Severity: conditions.Urgent, Phase: 2,
needs: func(f *signalFacts) error { return f.leaseErr }, watch: watchLease, needs: func(f *signalFacts) error { return f.leaseErr }, watch: watchLease,
newest: func(f *signalFacts) time.Time { return f.lease.renewed }}, newest: func(f *signalFacts) time.Time { return f.lease.renewed }},
@@ -644,10 +646,29 @@ func watchLease(f *signalFacts) []conditions.Observation {
"does not know it lost it", "does not know it lost it",
Said: l.resetSaid}) Said: l.resetSaid})
} }
// An epoch below the one this controller holds was taken over by it: said once by the takeover itself,
// raised and cleared (sayTakeover), not observed for an hour after (novox/hq issue 906). One that ended
// while this controller does not hold the lease — serving without it — is still said for the hour.
var ended []inventory.Epoch
for _, e := range l.ended {
if l.held && e.Epoch < l.epoch {
continue
}
ended = append(ended, e)
}
if o, ok := lostObservation(ended, l.epoch, f.now); ok {
out = append(out, o)
}
return out
}
// lostObservation is the lease-lost condition of the epochs that lost the lease or let it expire within the
// hour, named by the newest; false when none did. An epoch given back is no loss.
func lostObservation(ended []inventory.Epoch, acting uint64, now time.Time) (conditions.Observation, bool) {
var lost []string var lost []string
newest := inventory.Epoch{} newest := inventory.Epoch{}
for _, e := range l.ended { for _, e := range ended {
if e.Ended == nil || e.How == inventory.EpochReleased || f.now.Sub(*e.Ended) > advisoryQuiet { if e.Ended == nil || e.How == inventory.EpochReleased || now.Sub(*e.Ended) > advisoryQuiet {
continue continue
} }
lost = append(lost, fmt.Sprintf("epoch %d (%s) %s at %s", e.Epoch, e.Instance, e.How, lost = append(lost, fmt.Sprintf("epoch %d (%s) %s at %s", e.Epoch, e.Instance, e.How,
@@ -656,18 +677,95 @@ func watchLease(f *signalFacts) []conditions.Observation {
newest = e newest = e
} }
} }
if len(lost) > 0 { if len(lost) == 0 {
return conditions.Observation{}, false
}
how := "lost it: another holder wrote its key, or no renewal was made within the key's age" how := "lost it: another holder wrote its key, or no renewal was made within the key's age"
if newest.How == inventory.EpochExpired { if newest.How == inventory.EpochExpired {
how = "stopped renewing it without giving it back, and was taken over" how = "stopped renewing it without giving it back, and was taken over"
} }
out = append(out, conditions.Observation{Scope: conditions.ScopeCore, ID: "controller.lease", Token: "lost", return conditions.Observation{Scope: conditions.ScopeCore, ID: "controller.lease", Token: "lost",
Kind: "lease-lost", Machine: newest.Host, Severity: conditions.Urgent, Kind: "lease-lost", Machine: newest.Host, Severity: conditions.Urgent, Source: "S12",
Summary: fmt.Sprintf("the controller of epoch %d (%s) %s; the controller of epoch %d acts now", newest.Epoch, Summary: fmt.Sprintf("the controller of epoch %d (%s) %s; the controller of epoch %d acts now", newest.Epoch,
newest.Instance, how, l.epoch), newest.Instance, how, acting),
Said: strings.Join(lost, "; ")}) Said: strings.Join(lost, "; ")}, true
}
// takeover is the condition a takeover says: the epoch just before the one taken, when it lost the lease or
// let it expire; false when it gave it back, or none is recorded.
func takeover(ended []inventory.Epoch, taken uint64, now time.Time) (conditions.Observation, bool) {
var before *inventory.Epoch
for i, e := range ended {
if e.Epoch < taken && (before == nil || e.Epoch > before.Epoch) {
before = &ended[i]
}
}
if before == nil {
return conditions.Observation{}, false
}
return lostObservation([]inventory.Epoch{*before}, taken, now)
}
// sayTakeover is the takeover said as an event (novox/hq issue 906): when the controller before this one lost
// the lease or let it expire, the lease-lost condition is raised and cleared at once by this controller, which
// holds the lease — the operator reads that it happened and that it is over, not an urgent alarm for an hour.
// A lease given back says nothing. Said once the condition store is open; a take before that is said then.
func (a *actor) sayTakeover(ctx context.Context, epoch uint64) {
a.mu.Lock()
keeper, inv := a.keeper, a.inv
a.mu.Unlock()
if keeper == nil || inv == nil || epoch == 0 {
return
}
now := time.Now()
reading, cancel := context.WithTimeout(ctx, 10*time.Second)
defer cancel()
ended, err := inv.EpochsSince(reading, now.Add(-advisoryQuiet))
if err != nil {
// Failure is loud: S12 no longer observes an epoch below the one held, so an unread takeover is said as
// one that was not a handover — raised, and left for S12 to clear (ADR 0558 §3).
fmt.Printf("how the epoch before %d ended cannot be read, so its takeover is said as an expiry: %v\n", epoch, err)
o := conditions.Observation{Scope: conditions.ScopeCore, ID: "controller.lease", Token: "lost",
Kind: "lease-lost", Severity: conditions.Urgent, Source: "S12",
Summary: fmt.Sprintf("the controller of epoch %d took the lease, and how the controller before it ended "+
"cannot be read: it is said as one that stopped renewing without giving it back", epoch),
Said: err.Error()}
if _, err := keeper.Observe(reading, o); err != nil {
fmt.Printf("the takeover at epoch %d could not be said: %v\n", epoch, err)
}
return
}
sayTakeoverOf(reading, keeper, ended, epoch, now)
}
// sayTakeoverOf raises and clears the takeover's condition over the epochs read.
func sayTakeoverOf(ctx context.Context, keeper *conditions.Keeper, ended []inventory.Epoch, epoch uint64, now time.Time) {
o, ok := takeover(ended, epoch, now)
if !ok {
return
}
if _, err := keeper.Observe(ctx, o); err != nil {
fmt.Printf("the takeover at epoch %d could not be said: %v\n", epoch, err)
return
}
if _, err := keeper.Clear(ctx, o.Key(), fmt.Sprintf("taken over: the controller of epoch %d holds the lease", epoch)); err != nil {
fmt.Printf("the takeover at epoch %d was said and could not be cleared: %v\n", epoch, err)
}
}
// watchTakeovers gives the serving controller's condition store to the lease, and says the takeover of the
// epoch it holds, if it took one over before the store opened.
func (a *actor) watchTakeovers(ctx context.Context, inv *inventory.Inventory, keeper *conditions.Keeper) {
a.mu.Lock()
a.keeper, a.inv = keeper, inv
held := a.held
a.mu.Unlock()
if held == nil {
return
}
if epoch, err := held.Epoch(); err == nil {
a.sayTakeover(ctx, epoch)
} }
return out
} }
// handActsWithin is how far back a repeated cause counts (S15). // handActsWithin is how far back a repeated cause counts (S15).
+32 -1
View File
@@ -491,13 +491,41 @@ func TestAStaleWriterIsNamedByItsEpoch(t *testing.T) {
} }
} }
// **The lease lost is said for an hour, and serving without it for as long as it lasts** (S12). // **The lease lost is said for an hour while nobody took it over, a takeover is said once as it happens —
// raised and cleared by the controller that took it — a lease given back is never said, and serving without
// it for as long as it lasts** (S12, novox/hq issue 906).
func TestALeaseLostOrMissingIsSaid(t *testing.T) { func TestALeaseLostOrMissingIsSaid(t *testing.T) {
now := time.Now() now := time.Now()
f := calm(now) f := calm(now)
expired := now.Add(-59 * time.Minute) expired := now.Add(-59 * time.Minute)
f.lease.ended = []inventory.Epoch{{Epoch: 41, Instance: "controller@anchor pid 7", Host: "anchor", f.lease.ended = []inventory.Epoch{{Epoch: 41, Instance: "controller@anchor pid 7", Host: "anchor",
Ended: &expired, How: inventory.EpochExpired}} Ended: &expired, How: inventory.EpochExpired}}
// Held at epoch 57, above it: the takeover said it, and no hour of observation follows.
if got := watchLease(f); len(got) != 0 {
t.Fatalf("an epoch this controller took over is observed after the takeover: %+v", got)
}
store := conditions.NewInMemory()
k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store, Now: func() time.Time { return now }})
sayTakeoverOf(t.Context(), k, f.lease.ended, 57, now)
if open, err := k.Open(t.Context()); err != nil || len(open) != 0 {
t.Fatalf("the takeover is left open: %+v, %v", open, err)
}
k.Close(t.Context()) // its transitions written to the history
history, err := store.Since(t.Context(), now.Add(-time.Minute))
if err != nil {
t.Fatal(err)
}
var changes []string
for _, e := range history {
if e.Key == "core.controller.lease.lost" {
changes = append(changes, e.Change)
}
}
if !slices.Equal(changes, []string{"raised", "cleared"}) {
t.Fatalf("the takeover of an expired lease is not raised and cleared: %v", changes)
}
// Not holding the lease, the loss is said for the hour.
f.lease.held = false
got := watchLease(f) got := watchLease(f)
if len(got) != 1 || got[0].Key() != "core.controller.lease.lost" || !strings.Contains(got[0].Summary, "epoch 41") || if len(got) != 1 || got[0].Key() != "core.controller.lease.lost" || !strings.Contains(got[0].Summary, "epoch 41") ||
got[0].Severity != conditions.Urgent { got[0].Severity != conditions.Urgent {
@@ -512,6 +540,9 @@ func TestALeaseLostOrMissingIsSaid(t *testing.T) {
if got := watchLease(f); len(got) != 0 { if got := watchLease(f); len(got) != 0 {
t.Fatalf("a lease given back is said as lost: %+v", got) t.Fatalf("a lease given back is said as lost: %+v", got)
} }
if _, said := takeover(f.lease.ended, 57, now); said {
t.Fatal("the takeover of a lease given back is said")
}
f.lease = leaseFacts{held: true, epoch: 501, renewed: now, reset: now.Add(-time.Minute), resetSaid: "moved past 500"} f.lease = leaseFacts{held: true, epoch: 501, renewed: now, reset: now.Add(-time.Minute), resetSaid: "moved past 500"}
if got := watchLease(f); len(got) != 1 || got[0].Key() != "core.controller.lease.reset" { if got := watchLease(f); len(got) != 1 || got[0].Key() != "core.controller.lease.reset" {
t.Fatalf("a lease bucket raised again from nothing was not said: %+v", got) t.Fatalf("a lease bucket raised again from nothing was not said: %+v", got)
+1 -1
View File
@@ -3,7 +3,7 @@ module github.com/novox/mesh-controller
go 1.26.0 go 1.26.0
require ( require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024234636-10b6e202320f git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026220044-f178323b5790
github.com/jackc/pgx/v5 v5.10.0 github.com/jackc/pgx/v5 v5.10.0
github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0 github.com/nats-io/nats.go v1.54.0
+2 -2
View File
@@ -1,7 +1,7 @@
git.novox.be/novox/mesh-host v0.0.0-20261018171808-9d9e3a46c4f5 h1:E5stfIeCN6uYgA9TCv+4K8JgaURg3zQppvUgJLaDMP4= git.novox.be/novox/mesh-host v0.0.0-20261018171808-9d9e3a46c4f5 h1:E5stfIeCN6uYgA9TCv+4K8JgaURg3zQppvUgJLaDMP4=
git.novox.be/novox/mesh-host v0.0.0-20261018171808-9d9e3a46c4f5/go.mod h1:4wyoCr6IIIIarEUuROThWDj7DAvJvIidqMjWcpSnOwE= git.novox.be/novox/mesh-host v0.0.0-20261018171808-9d9e3a46c4f5/go.mod h1:4wyoCr6IIIIarEUuROThWDj7DAvJvIidqMjWcpSnOwE=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024234636-10b6e202320f h1:lUEyNPPF3w4Q0LIgcUUD6ofkKVhCF0n7dVjMoixQmXo= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026220044-f178323b5790 h1:WB95Hi2VW+OL6Nz+G3G1Vg4NoICUmyTA/PiC+U2mgLM=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024234636-10b6e202320f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026220044-f178323b5790/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+9 -7
View File
@@ -46,9 +46,8 @@ func TestAConsumerMadeFromNowNeverReplaysTheStreamsHistory(t *testing.T) {
t.Fatalf("a consumer made from now holds %d of the stream's past (policy %v)", info.NumPending, info.Config.DeliverPolicy) t.Fatalf("a consumer made from now holds %d of the stream's past (policy %v)", info.NumPending, info.Config.DeliverPolicy)
} }
_, _ = js.js.Publish("history.merged", []byte("new")) _, _ = js.js.Publish("history.merged", []byte("new"))
time.Sleep(100 * time.Millisecond) if got := pendingOf(js, stream, "fresh", 1); got != 1 {
if info, _ = js.js.ConsumerInfo(stream, "fresh"); info.NumPending != 1 { t.Fatalf("a new announcement is not pending: %d", got)
t.Fatalf("a new announcement is not pending: %d", info.NumPending)
} }
// Asserted again, it keeps where it is. // Asserted again, it keeps where it is.
if err := js.EnsureConsumer(fresh); err != nil { if err := js.EnsureConsumer(fresh); err != nil {
@@ -61,8 +60,8 @@ func TestAConsumerMadeFromNowNeverReplaysTheStreamsHistory(t *testing.T) {
if err := js.EnsureConsumer(old); err != nil { if err := js.EnsureConsumer(old); err != nil {
t.Fatal(err) t.Fatal(err)
} }
if info, _ = js.js.ConsumerInfo(stream, "old"); info.NumPending != 6 { if got := pendingOf(js, stream, "old", 6); got != 6 {
t.Fatalf("the default should replay all six: %d", info.NumPending) t.Fatalf("the default should replay all six: %d", got)
} }
before, after, err := js.ResetConsumer(stream, "old") before, after, err := js.ResetConsumer(stream, "old")
if err != nil { if err != nil {
@@ -89,10 +88,13 @@ func TestAConsumerMadeFromNowNeverReplaysTheStreamsHistory(t *testing.T) {
} }
} }
// pendingOf polls a consumer's pending count until it is want or a second has passed, and returns what it saw last. // pendingOf polls a consumer's pending count until it is want or ten seconds have passed, and returns what it saw
// last. The server counts a consumer's pending after it acknowledges the publish, not before, so a test that reads
// the count at once fails on a loaded build node (a try of main on ace, 2026-10-26): every test that expects a
// count after a publish reads it through here.
func pendingOf(js *JetStream, stream, name string, want uint64) uint64 { func pendingOf(js *JetStream, stream, name string, want uint64) uint64 {
var got uint64 var got uint64
for deadline := time.Now().Add(time.Second); ; time.Sleep(20 * time.Millisecond) { for deadline := time.Now().Add(10 * time.Second); ; time.Sleep(20 * time.Millisecond) {
if info, err := js.js.ConsumerInfo(stream, name); err == nil { if info, err := js.js.ConsumerInfo(stream, name); err == nil {
got = info.NumPending got = info.NumPending
} }
@@ -73,15 +73,15 @@ func TestAResetOfTheEventsConsumerLeavesTheMergesAndTheirConsumerAlone(t *testin
if _, err := js.Context().Publish(ControllerFollows[0], []byte(`{}`)); err != nil { if _, err := js.Context().Publish(ControllerFollows[0], []byte(`{}`)); err != nil {
t.Fatal(err) t.Fatal(err)
} }
merges, err := js.Context().ConsumerInfo(EventsStream, MergesConsumerName) if got := pendingOf(js, EventsStream, MergesConsumerName, 1); got != 1 {
if err != nil || merges.NumPending != 1 { t.Fatalf("the merges consumer holds %d pending; want the merge", got)
t.Fatalf("the merges consumer holds %v (%v); want the merge", merges, err)
} }
if _, _, err := js.ResetConsumer(EventsStream, ControllerName); err != nil { if _, _, err := js.ResetConsumer(EventsStream, ControllerName); err != nil {
t.Fatal(err) t.Fatal(err)
} }
merges, err = js.Context().ConsumerInfo(EventsStream, MergesConsumerName) pending := pendingOf(js, EventsStream, MergesConsumerName, 1)
if err != nil || merges.NumPending != 1 || merges.Delivered.Stream != 0 { merges, err := js.Context().ConsumerInfo(EventsStream, MergesConsumerName)
if err != nil || pending != 1 || merges.Delivered.Stream != 0 {
t.Fatalf("after the reset the merges consumer holds %+v (%v); want its merge left alone", merges, err) t.Fatalf("after the reset the merges consumer holds %+v (%v); want its merge left alone", merges, err)
} }
} }
@@ -107,9 +107,8 @@ func TestAMergeTheOldConsumerHadNotAcknowledgedIsTheMergesConsumers(t *testing.T
if err := AssertMeshConsumers(js); err != nil { if err := AssertMeshConsumers(js); err != nil {
t.Fatal(err) t.Fatal(err)
} }
merges, err := js.Context().ConsumerInfo(EventsStream, MergesConsumerName) if got := pendingOf(js, EventsStream, MergesConsumerName, 1); got != 1 {
if err != nil || merges.NumPending != 1 { t.Fatalf("the merges consumer holds %d pending; want the merge the old consumer never acknowledged", got)
t.Fatalf("the merges consumer holds %v (%v); want the merge the old consumer never acknowledged", merges, err)
} }
} }
+2 -3
View File
@@ -52,9 +52,8 @@ func TestAModuleStreamOnARealBus(t *testing.T) {
t.Fatalf("publishing %s's mail: %v", account, err) t.Fatalf("publishing %s's mail: %v", account, err)
} }
} }
time.Sleep(50 * time.Millisecond) if got := pendingOf(js, "MAIL", c.Name, 1); got != 1 {
if info, _ := js.js.ConsumerInfo("MAIL", c.Name); info.NumPending != 1 { t.Fatalf("the reader granted `mesh` has %d pending, want 1", got)
t.Fatalf("the reader granted `mesh` has %d pending, want 1", info.NumPending)
} }
narrowed := c narrowed := c
narrowed.Filters = []string{"mesh.stream.mesh-mail.event.mail.mesh.message.changed"} narrowed.Filters = []string{"mesh.stream.mesh-mail.event.mail.mesh.message.changed"}
+20 -2
View File
@@ -24,8 +24,9 @@ import (
// subjects cover it, so what travels there — a code typed by the operator — is never persisted. A kinded // subjects cover it, so what travels there — a code typed by the operator — is never persisted. A kinded
// holder asks with its own kind; the modules that watch the seat answer. // holder asks with its own kind; the modules that watch the seat answer.
// //
// And one read: **a holder's records**, a bucket the seat names, read by each user under its own name only // And two reads: **a holder's records**, a bucket the seat names, read by each user under its own name only
// (`$KV.<bucket>.<module>.>`), so an asker reads the state of its own asks and no other asker's. // (`$KV.<bucket>.<module>.>`), so an asker reads the state of its own asks and no other asker's; and on a seat pulled by
// kind, **what its workers gave up on**, the copies DEAD_LETTERS keeps of the seat's stream (DeadLetterReads, issue 734).
// BuildAskMaxDeliver bounds the deliveries of a build seat's ask (issue 1907): mesh-sdk delivery.AskMaxDeliver, written // BuildAskMaxDeliver bounds the deliveries of a build seat's ask (issue 1907): mesh-sdk delivery.AskMaxDeliver, written
// here as the broker mirrors the catalogue rather than importing it. // here as the broker mirrors the catalogue rather than importing it.
@@ -129,6 +130,9 @@ func SeatTrafficOf(module string, holds, uses, watches []Seat) SeatTraffic {
} }
t.Records = append(t.Records, "$JS.API.DIRECT.GET.KV_"+b+".$KV."+b+"."+module+".>") t.Records = append(t.Records, "$JS.API.DIRECT.GET.KV_"+b+".$KV."+b+"."+module+".>")
} }
if s.PulledByKind && len(s.Accepts) > 0 && safeSubject.MatchString(s.Name) {
t.Records = append(t.Records, DeadLetterReads(seatStreamName(s.Name))...)
}
} }
for _, w := range watches { for _, w := range watches {
if w.Kinded { if w.Kinded {
@@ -151,6 +155,20 @@ func SeatTrafficOf(module string, holds, uses, watches []Seat) SeatTraffic {
return t return t
} }
// DeadLetterReads is what a user of a seat pulled by kind is granted to hear which of the asks it put the seat's workers
// gave up on (novox/hq issue 734): an ordered consumer on DEAD_LETTERS filtered to the copies the controller keeps of that
// seat's stream (DeadLetterSubject(stream, *)), pulled, and its deletion. A copy carries the ask's own message id
// (Mesh-Dead-Msg-Id), so the asker ends the step it waits for instead of waiting for ever; it reads the seat's dead
// letters and no other stream's, and changes none.
func DeadLetterReads(stream string) []string {
return []string{
"$JS.API.STREAM.INFO." + DeadLettersStream,
"$JS.API.CONSUMER.CREATE." + DeadLettersStream + ".*." + DeadLetterSubject(stream, ">"),
"$JS.API.CONSUMER.MSG.NEXT." + DeadLettersStream + ".*",
"$JS.API.CONSUMER.DELETE." + DeadLettersStream + ".*",
}
}
// seatTrafficGrants is the bus permissions seat traffic needs: the subjects themselves, and the JetStream API a // seatTrafficGrants is the bus permissions seat traffic needs: the subjects themselves, and the JetStream API a
// worker is pulled and acknowledged through and a record is read through. // worker is pulled and acknowledged through and a record is read through.
func seatTrafficGrants(t SeatTraffic) (pub, sub []string) { func seatTrafficGrants(t SeatTraffic) (pub, sub []string) {
+13
View File
@@ -403,6 +403,19 @@ func TestTheBuilderMayAskABuildSeatAndASessionMayNot(t *testing.T) {
t.Errorf("the builder may not publish %s", s) t.Errorf("the builder may not publish %s", s)
} }
} }
// Issue 734: the asker hears which of its asks the seat's workers gave up on, from DEAD_LETTERS, and no other stream's.
for subject, want := range map[string]bool{
"$JS.API.CONSUMER.CREATE.DEAD_LETTERS.x1.mesh.events.dead.SEAT_NODE_BUILD_GO.>": true,
"$JS.API.CONSUMER.DELETE.DEAD_LETTERS.x1": true,
"$JS.API.CONSUMER.MSG.NEXT.DEAD_LETTERS.x1": true,
"$JS.API.CONSUMER.CREATE.DEAD_LETTERS.x1.mesh.events.dead.EVENTS.>": false,
"$JS.API.STREAM.PURGE.DEAD_LETTERS": false,
"$JS.FC.DEAD_LETTERS.x1": false,
} {
if got := allowed(builder.Publish, subject); got != want {
t.Errorf("the builder may publish %s: %v, want %v", subject, got, want)
}
}
session := perms(t, Principal{Kind: KindPerson, Module: "jochen", Invokes: []string{"mesh-controller.status"}}) session := perms(t, Principal{Kind: KindPerson, Module: "jochen", Invokes: []string{"mesh-controller.status"}})
if allowed(session.Publish, "mesh.seat.node-build-go.accept.delivery") { if allowed(session.Publish, "mesh.seat.node-build-go.accept.delivery") {
t.Error("a session's account may put an ask") t.Error("a session's account may put an ask")
+12 -3
View File
@@ -67,14 +67,23 @@ func (i *Inventory) HighestEpoch(ctx context.Context) (uint64, error) {
} }
// TookEpoch records an epoch taken, and ends every earlier one still open as found expired: the lease // TookEpoch records an epoch taken, and ends every earlier one still open as found expired: the lease
// was free to take, so whoever held it last neither holds it nor said it let go. Answers the epochs it // was free to take, so whoever held it last neither holds it nor said it let go. givenBack is the epoch the
// ended that way, newest first. // bus says gave the key back (the lease's own delete marker names it, novox/hq issue 906, ADR 0558 §3): that
func (i *Inventory) TookEpoch(ctx context.Context, e Epoch) ([]Epoch, error) { // epoch, if still open, is ended as released — its holder gave the key back and had not yet recorded it — and
// every other as expired; zero is none. Answers the epochs it ended as expired, newest first.
func (i *Inventory) TookEpoch(ctx context.Context, e Epoch, givenBack uint64) ([]Epoch, error) {
tx, err := i.store.Pool().Begin(ctx) tx, err := i.store.Pool().Begin(ctx)
if err != nil { if err != nil {
return nil, err return nil, err
} }
defer func() { _ = tx.Rollback(ctx) }() defer func() { _ = tx.Rollback(ctx) }()
if givenBack != 0 && givenBack < e.Epoch {
if _, err := tx.Exec(ctx,
`update controller_epoch set ended = now(), how = $2 where epoch = $1 and ended is null`,
int64(givenBack), EpochReleased); err != nil {
return nil, err
}
}
rows, err := tx.Query(ctx, rows, err := tx.Query(ctx,
`update controller_epoch set ended = now(), how = $2 `update controller_epoch set ended = now(), how = $2
where ended is null and epoch < $1 where ended is null and epoch < $1
+71 -9
View File
@@ -34,9 +34,11 @@ import (
"encoding/json" "encoding/json"
"errors" "errors"
"fmt" "fmt"
"strconv"
"sync" "sync"
"time" "time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream" "github.com/nats-io/nats.go/jetstream"
) )
@@ -115,6 +117,7 @@ type Options struct {
// Lease is one instance's hold, or its wait for one. // Lease is one instance's hold, or its wait for one.
type Lease struct { type Lease struct {
kv jetstream.KeyValue kv jetstream.KeyValue
js jetstream.JetStream
stream jetstream.Stream stream jetstream.Stream
ttl time.Duration ttl time.Duration
o Options o Options
@@ -139,6 +142,9 @@ type Lease struct {
lostWhy error lostWhy error
lost chan struct{} lost chan struct{}
taken time.Time taken time.Time
// givenBack is the epoch whose holder gave back the key this instance took — the epoch its delete marker
// names on the bus — and zero when the key was left to expire, or freed by anything else (novox/hq issue 906).
givenBack uint64
// update writes the key at a revision: the bucket's own, or a test's bus that does not answer. // update writes the key at a revision: the bucket's own, or a test's bus that does not answer.
update func(ctx context.Context, key string, value []byte, last uint64) (uint64, error) update func(ctx context.Context, key string, value []byte, last uint64) (uint64, error)
} }
@@ -184,7 +190,7 @@ func Open(ctx context.Context, js jetstream.JetStream, bucket string, o Options)
return nil, fmt.Errorf("a lease renewed every %s with a margin of %s does not fit a key that lives %s", return nil, fmt.Errorf("a lease renewed every %s with a margin of %s does not fit a key that lives %s",
o.RenewEvery, o.Margin, status.TTL()) o.RenewEvery, o.Margin, status.TTL())
} }
return &Lease{kv: kv, stream: stream, ttl: status.TTL(), o: o, lost: make(chan struct{}), update: kv.Update}, nil return &Lease{kv: kv, js: js, stream: stream, ttl: status.TTL(), o: o, lost: make(chan struct{}), update: kv.Update}, nil
} }
// TTL is how long the key lives unrenewed. // TTL is how long the key lives unrenewed.
@@ -271,6 +277,7 @@ func (l *Lease) TryTake(ctx context.Context) (uint64, error) {
} else if found { } else if found {
return 0, &heldBy{h} return 0, &heldBy{h}
} }
givenBack := l.wasGivenBack(ctx)
if err := l.aboveTheFloor(ctx); err != nil { if err := l.aboveTheFloor(ctx); err != nil {
return 0, err return 0, err
} }
@@ -307,13 +314,51 @@ func (l *Lease) TryTake(ctx context.Context) (uint64, error) {
"back to be taken again", revision, floor) "back to be taken again", revision, floor)
} }
l.mu.Lock() l.mu.Lock()
l.held, l.epoch, l.revision, l.taken = true, revision, revision, now l.held, l.epoch, l.revision, l.taken, l.givenBack = true, revision, revision, now, givenBack
l.renewedAt(anchor) l.renewedAt(anchor)
l.mu.Unlock() l.mu.Unlock()
l.o.Say("took the controller lease at epoch %d (%s)", revision, h.Instance) l.o.Say("took the controller lease at epoch %d (%s)", revision, h.Instance)
return revision, nil return revision, nil
} }
// ReleasedHeader is the header of the delete marker a holder's Release writes, naming the epoch it gave back:
// the one mark that says a handover. A delete marker without it — a command's own lease given back, a take
// given back over the floor — names no epoch, so it says nothing about the holder before (novox/hq issue 906).
const ReleasedHeader = "Mesh-Lease-Released"
// wasGivenBack is the epoch the free key was given back by: the epoch its last message on the bus — the delete
// marker a holder's Release writes — names. Zero when the key expired (the bucket's age removed every message),
// when the marker names no epoch, and when it cannot be read: a takeover nobody can prove was a handover is
// said as one that was not (novox/hq issue 906, ADR 0558 §3).
func (l *Lease) wasGivenBack(ctx context.Context) uint64 {
if l.stream == nil {
return 0
}
last, err := l.stream.GetLastMsgForSubject(ctx, l.subject())
if err != nil || last == nil {
return 0
}
if op := last.Header.Get("KV-Operation"); op != "DEL" && op != "PURGE" {
return 0
}
epoch, err := strconv.ParseUint(last.Header.Get(ReleasedHeader), 10, 64)
if err != nil {
return 0
}
return epoch
}
// subject is the lease key's subject in its bucket's stream.
func (l *Lease) subject() string { return "$KV." + l.kv.Bucket() + "." + Key }
// GivenBack is the epoch whose holder gave back the key this instance took, zero when it was not given back:
// what the takeover was, read from the bus (novox/hq issue 906).
func (l *Lease) GivenBack() uint64 {
l.mu.Lock()
defer l.mu.Unlock()
return l.givenBack
}
// floor is the highest epoch issued, zero without a floor. // floor is the highest epoch issued, zero without a floor.
func (l *Lease) floor(ctx context.Context) (uint64, error) { func (l *Lease) floor(ctx context.Context) (uint64, error) {
if l.o.Floor == nil { if l.o.Floor == nil {
@@ -588,9 +633,11 @@ func (l *Lease) lose(why error) error {
return l.lostWhy return l.lostWhy
} }
// Release gives the lease back, so the next candidate takes it at once rather than after its age. // Release gives the lease back, so the next candidate takes it at once rather than after its age, and answers
// Only the key this instance last wrote is deleted: one that moved is somebody else's. // the epoch it gave back — zero when it gave nothing back: it held none, had lost it, or the bus would not
func (l *Lease) Release(ctx context.Context) { // delete the key. Only the key this instance last wrote is deleted: one that moved is somebody else's. The
// gate closes before the key goes, so nothing this instance starts after overlaps its successor.
func (l *Lease) Release(ctx context.Context) uint64 {
l.mu.Lock() l.mu.Lock()
held, revision, epoch := l.held && l.lostWhy == nil, l.revision, l.epoch held, revision, epoch := l.held && l.lostWhy == nil, l.revision, l.epoch
l.held = false l.held = false
@@ -600,11 +647,11 @@ func (l *Lease) Release(ctx context.Context) {
} }
l.mu.Unlock() l.mu.Unlock()
if !held { if !held {
return return 0
} }
releasing, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) releasing, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
defer cancel() defer cancel()
err := l.kv.Delete(releasing, Key, jetstream.LastRevision(revision)) err := l.giveBack(releasing, epoch, revision)
if err != nil && isWrongLast(err) { if err != nil && isWrongLast(err) {
// A renewal was in flight as this began and landed first: the key moved, and is still this // A renewal was in flight as this began and landed first: the key moved, and is still this
// instance's if it still names it at this epoch. Deleted at the revision it has now. // instance's if it still names it at this epoch. Deleted at the revision it has now.
@@ -612,14 +659,29 @@ func (l *Lease) Release(ctx context.Context) {
var h Holder var h Holder
if json.Unmarshal(entry.Value(), &h) == nil && h.Instance == l.o.Holder.Instance && if json.Unmarshal(entry.Value(), &h) == nil && h.Instance == l.o.Holder.Instance &&
(h.Epoch == epoch || h.Epoch == 0 && entry.Revision() == epoch) { (h.Epoch == epoch || h.Epoch == 0 && entry.Revision() == epoch) {
err = l.kv.Delete(releasing, Key, jetstream.LastRevision(entry.Revision())) err = l.giveBack(releasing, epoch, entry.Revision())
} }
} }
} }
if err != nil { if err != nil {
l.o.Say("the controller lease (epoch %d) could not be given back, so the next controller waits for "+ l.o.Say("the controller lease (epoch %d) could not be given back, so the next controller waits for "+
"it to expire: %v", epoch, err) "it to expire: %v", epoch, err)
return return 0
} }
l.o.Say("gave the controller lease back (epoch %d)", epoch) l.o.Say("gave the controller lease back (epoch %d)", epoch)
return epoch
}
// giveBack deletes the key at the revision this instance last wrote, as the bucket's own delete does — a
// marker with KV-Operation DEL, compare-and-set on the key's last revision — and names the epoch given back in
// the marker, so the next holder can tell this handover from any other delete (ReleasedHeader).
func (l *Lease) giveBack(ctx context.Context, epoch, revision uint64) error {
if l.js == nil {
return l.kv.Delete(ctx, Key, jetstream.LastRevision(revision))
}
msg := nats.NewMsg(l.subject())
msg.Header.Set("KV-Operation", "DEL")
msg.Header.Set(ReleasedHeader, strconv.FormatUint(epoch, 10))
_, err := l.js.PublishMsg(ctx, msg, jetstream.WithExpectLastSequencePerSubject(revision))
return err
} }
+17 -3
View File
@@ -133,7 +133,12 @@ func TestTheSecondControllerWaitsAndTakesAHigherEpochOnHandover(t *testing.T) {
// a gives it back: b takes it at once, not after the key's age, at a higher epoch. // a gives it back: b takes it at once, not after the key's age, at a higher epoch.
stop() stop()
handedOver := time.Now() handedOver := time.Now()
a.Release(t.Context()) if given := a.Release(t.Context()); given != epochA {
t.Fatalf("a's release answers epoch %d given back, want %d", given, epochA)
}
if again := a.Release(t.Context()); again != 0 {
t.Fatalf("a second release gives epoch %d back again", again)
}
select { select {
case epochB := <-took: case epochB := <-took:
if epochB <= epochA { if epochB <= epochA {
@@ -142,6 +147,9 @@ func TestTheSecondControllerWaitsAndTakesAHigherEpochOnHandover(t *testing.T) {
if waited := time.Since(handedOver); waited > testTTL { if waited := time.Since(handedOver); waited > testTTL {
t.Fatalf("b took the lease %s after a gave it back: it waited out the key's age", waited) t.Fatalf("b took the lease %s after a gave it back: it waited out the key's age", waited)
} }
if given := b.GivenBack(); given != epochA {
t.Fatalf("b reads from the bus that epoch %d gave the key back, want a's %d (novox/hq issue 906)", given, epochA)
}
case <-time.After(2 * testTTL): case <-time.After(2 * testTTL):
t.Fatal("b never took the lease a gave back") t.Fatal("b never took the lease a gave back")
} }
@@ -182,6 +190,9 @@ func TestAHolderThatStopsRenewingStopsActingBeforeItIsTakenOver(t *testing.T) {
if epochB <= epochA { if epochB <= epochA {
t.Fatalf("b took epoch %d after a's %d", epochB, epochA) t.Fatalf("b took epoch %d after a's %d", epochB, epochA)
} }
if given := b.GivenBack(); given != 0 {
t.Fatalf("b reads a key that expired as given back by epoch %d (novox/hq issue 906)", given)
}
// And a, the moment it tries to renew, knows it lost: the key moved. // And a, the moment it tries to renew, knows it lost: the key moved.
if err := a.Renew(t.Context()); err == nil { if err := a.Renew(t.Context()); err == nil {
t.Fatal("a renewed a lease b holds") t.Fatal("a renewed a lease b holds")
@@ -226,8 +237,11 @@ func TestARenewalRefusedIsALossAtOnce(t *testing.T) {
if _, err := a.Epoch(); !errors.Is(err, ErrNotHeld) { if _, err := a.Epoch(); !errors.Is(err, ErrNotHeld) {
t.Fatalf("a acts after its renewal was refused: %v", err) t.Fatalf("a acts after its renewal was refused: %v", err)
} }
// And giving back a lease it lost deletes nothing of the one who holds it now. // And giving back a lease it lost deletes nothing of the one who holds it now, and answers no epoch given
a.Release(t.Context()) // back, so none is recorded released (novox/hq issue 906).
if given := a.Release(t.Context()); given != 0 {
t.Fatalf("a says it gave back epoch %d of a lease it lost", given)
}
if h, found, err := Current(t.Context(), kv); err != nil || !found || h.Instance != "intruder" { if h, found, err := Current(t.Context(), kv); err != nil || !found || h.Instance != "intruder" {
t.Fatalf("after a gave back what it lost, the key holds %+v (%v, %v)", h, found, err) t.Fatalf("after a gave back what it lost, the key holds %+v (%v, %v)", h, found, err)
} }
+1
View File
@@ -82,6 +82,7 @@
"dead-letters", "dead-letters",
"root-free", "root-free",
"judge-directories", "judge-directories",
"placement-export",
"data", "data",
"build", "build",
"artifacts", "artifacts",
+1 -1
View File
@@ -9,7 +9,7 @@ a second place to clean besides the catalogue itself.
mesh-catalog 3a1017f5594217092a654dd1e2d72226766bc16a modules/*/module.json, modules/*/files, modules/*/templates, predefined.json mesh-catalog 3a1017f5594217092a654dd1e2d72226766bc16a modules/*/module.json, modules/*/files, modules/*/templates, predefined.json
mesh-host 9d9e3a46c4f5943cfffb413d962ce48b75e2fd39 examples/foundation-first-node.lock mesh-host 9d9e3a46c4f5943cfffb413d962ce48b75e2fd39 examples/foundation-first-node.lock
mesh-sdk 10b6e202320f38301c67a6822b0734e0b870ea30 conformance/events/module-event.json mesh-sdk f178323b57906eee876bf4e1ab832f08e8be6490 conformance/events/module-event.json
To move them, from this repository's root, with the two repositories checked out beside it: To move them, from this repository's root, with the two repositories checked out beside it: