mesh-grants holds the seat mesh-grants and writes the controller's lease on the operator's warrant
ADR 0582: lease-write is refused, naming each check, until the lease check at phase write and every condition of clause 2 read green (design 47's earlier issues resolved, every node trusting the vault's key, mesh-config's compare empty from the indexer, the upgrade compare empty, the keeper's take-over done, the vault minting with nothing held back, the bus module listing its expiry and removal tools). It asks the operator once; the yes binds the act, the holder, the ask and the checks' digest, and the checks are read again on the warrant (one that reads otherwise refuses it, named). Then, each step read back: node-service-manager.stop on the controller's node, the list composed with Records.LeaseCut and waited on until the bus node's have says it loaded it, the expiry removed through nats_bucket_expiry, the key holder written (with the ask and the time), lease-handed said, the act recorded through mesh-hand-acts.warranted, and removal-due written with a day to live. lease-altered is said once when the key or the expiry differs; lease-hand-back reverses it for a day and is refused once the removal ran; removal-due's marker alone starts the removal of the controller's consumer and the stream ASSIGNMENTS, recorded as hand acts, and a failure says removal-failed and writes the key again with a day (grants_removal_due is the recovery for a missed marker). From the write memberships are written to the state alone (ADR 0571 clause 10), the user-list key says lease-handed, and the user list's hand-back answers gone (ADR 0547 §11). Every step waits on an event, never a timer. Bent rule: S14-S21 are one commit, not eight: the steps are one state machine in leasewrite.go, and splitting it by hunks would review the same file eight times for nothing it protects. Tests: TestTheLeaseWrite (new, one function with a subtest per subtask), paid by the deletion of TestTheKeepListNamesItsRecordsAndThreeToTheWrite (moved to mesh-sdk, S4/S30) and the assigner's two deletions (S25). Fixtures shaped from each writer's code: systemd's Manager.Status, nats-tools' BucketExpirySet, ConsumerRemoved and StreamRemoved, mesh-issues' Brief, mesh-delivery-view's TrustsAnswer, mesh-config's Comparison, the assigner's upgrade compare, mesh-vault's vault_status. (issue 774, 774/S14, 774/S15, 774/S16, 774/S17, 774/S18, 774/S19, 774/S20, 774/S21) Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
This commit is contained in:
@@ -12,6 +12,7 @@ import (
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
)
|
||||
|
||||
// The three bus adapters for mesh-sdk go/asker on the part's own credential, as mesh-delivery-teller has them (copied: the
|
||||
@@ -81,3 +82,16 @@ func sayVia(ctx context.Context, js jetstream.JetStream, module string) func(c a
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// seatSaying says the seat mesh-grants' events on its seat subject into EVENTS, as the module that holds it (novox/hq
|
||||
// ADR 0582 clauses 3, 4, 6): lease-handed, lease-altered and removal-failed.
|
||||
func seatSaying(ctx context.Context, js jetstream.JetStream) func(event string, body any) error {
|
||||
return func(event string, body any) error {
|
||||
data, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_, err = js.Publish(ctx, "mesh.seat."+catalogue.GrantsSeat+".event."+event, data)
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,79 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
)
|
||||
|
||||
// leaseHand is the process's hand on the controller's lease (novox/hq ADR 0582 clause 3): its key holder, which the
|
||||
// lease write's cut grants the part alone, and the bucket's expiry, read from the bucket's own stream.
|
||||
type leaseHand struct {
|
||||
ctx context.Context
|
||||
js jetstream.JetStream
|
||||
|
||||
mu sync.Mutex
|
||||
kv jetstream.KeyValue
|
||||
}
|
||||
|
||||
func (l *leaseHand) bucket() (jetstream.KeyValue, error) {
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
if l.kv != nil {
|
||||
return l.kv, nil
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(l.ctx, requestWithin)
|
||||
defer cancel()
|
||||
kv, err := l.js.KeyValue(ctx, buscompose.LeaseBucket)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the lease's bucket cannot be opened: %w", err)
|
||||
}
|
||||
l.kv = kv
|
||||
return kv, nil
|
||||
}
|
||||
|
||||
func (l *leaseHand) PutHolder(value []byte) error {
|
||||
kv, err := l.bucket()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(l.ctx, requestWithin)
|
||||
defer cancel()
|
||||
_, err = kv.Put(ctx, buscompose.ControllerLeaseKey, value)
|
||||
return err
|
||||
}
|
||||
|
||||
func (l *leaseHand) DeleteHolder() error {
|
||||
kv, err := l.bucket()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(l.ctx, requestWithin)
|
||||
defer cancel()
|
||||
return kv.Delete(ctx, buscompose.ControllerLeaseKey)
|
||||
}
|
||||
|
||||
func (l *leaseHand) Expiry() (time.Duration, error) {
|
||||
kv, err := l.bucket()
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(l.ctx, requestWithin)
|
||||
defer cancel()
|
||||
st, err := kv.Status(ctx)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return st.TTL(), nil
|
||||
}
|
||||
|
||||
// PutTTL writes a key of the part's own state with its own time to live (the state lease).
|
||||
func (s ownStore) PutTTL(state, key string, value []byte, ttl time.Duration) (uint64, error) {
|
||||
return s.c.State(state).RenewTTL(key, json.RawMessage(value), ttl)
|
||||
}
|
||||
@@ -16,10 +16,8 @@
|
||||
// It holds no timer, and never exits for a refused read, write or publish: a refusal is said and tried again at the next
|
||||
// event.
|
||||
//
|
||||
// **No seat in this milestone.** ADR 0547 clause 5 gives the part the seat `mesh-grants`; a `mesh-*` seat must be in
|
||||
// mesh-sdk's own seat list, which the running controller's composer does not know, so the part serves module tools
|
||||
// (grants_*) and claims no seat until take-over. The composer grants the membership publish by the module's name,
|
||||
// mesh-grants.
|
||||
// **The seat mesh-grants** (novox/hq ADR 0582 clause 1, issue 774): the part holds it for the lease write, serving
|
||||
// lease-write and lease-hand-back beside its own grants_* tools, and says lease-handed, lease-altered and removal-failed.
|
||||
//
|
||||
// Started by the tool runner, with no credential of its own, it serves those tools (a process never serves, ADR 0436).
|
||||
package main
|
||||
@@ -121,6 +119,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
client := &asker.Client{Module: account.Module, State: conn.State(grants.StateTakeover), Publish: publishVia(ctx, js),
|
||||
Record: recordVia(ctx, nc, account.Module), Subscribe: decidedVia(nc, account.Module, logger.Printf)}
|
||||
f.Part.Asker, f.Part.Say = client, sayVia(ctx, js, account.Module)
|
||||
// The lease write's hand on the lease, and its seat's events (ADR 0582 clauses 3, 4, 6).
|
||||
f.Part.Lease, f.Part.SaySeat = &leaseHand{ctx: ctx, js: js}, seatSaying(ctx, js)
|
||||
f.Part.Forget = func(id string) error { return conn.State(grants.StateTakeover).Delete("ask." + id) }
|
||||
if err := client.Listen(f.Part.Warranted, f.Part.Said); err != nil {
|
||||
logger.Printf("the operator's answers cannot be heard yet, so take-over and hand-back wait: %v", err)
|
||||
|
||||
@@ -3,8 +3,10 @@ package main
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
|
||||
"mesh-grants/internal/grants"
|
||||
)
|
||||
@@ -44,6 +46,10 @@ func (runtimeStore) Keys(state string) ([]string, error) { return stdio.State(st
|
||||
|
||||
func (runtimeStore) Delete(state, key string) error { return stdio.State(state).Delete(key) }
|
||||
|
||||
func (runtimeStore) PutTTL(state, key string, value []byte, ttl time.Duration) (uint64, error) {
|
||||
return stdio.State(state).PutTTL(key, json.RawMessage(value), ttl)
|
||||
}
|
||||
|
||||
// runtimeBus calls tools and says events through the runtime.
|
||||
type runtimeBus struct{}
|
||||
|
||||
@@ -171,9 +177,40 @@ func toolsOf(p *grants.Part) []stdio.Tool {
|
||||
"with the tool that answers each. Asks the controller nothing; writes nothing. Read-only.",
|
||||
Input: schema(nil),
|
||||
Run: func(map[string]any) (any, error) { return p.LeaseCheck() }},
|
||||
// The seat's two verbs (ADR 0582 clauses 1, 5): asked of the operator by the part's process.
|
||||
{Name: grants.SeatTool(grants.ActLeaseWrite), Description: seatVerb(grants.ActLeaseWrite).Description,
|
||||
Input: seatVerb(grants.ActLeaseWrite).Input,
|
||||
Run: func(a map[string]any) (any, error) {
|
||||
return p.LeaseWrite(stdio.CallerOf(a), grants.Answer(a, "why"))
|
||||
}},
|
||||
{Name: grants.SeatTool(grants.ActLeaseHandBack), Description: seatVerb(grants.ActLeaseHandBack).Description,
|
||||
Input: seatVerb(grants.ActLeaseHandBack).Input,
|
||||
Run: func(a map[string]any) (any, error) {
|
||||
return p.LeaseHandBack(stdio.CallerOf(a), grants.Answer(a, "why"))
|
||||
}},
|
||||
{Name: grants.ToolRemovalDue, Description: "Write the lease write's removal-due again with a short time to live, so " +
|
||||
"its expiry marker starts the day-later removal of the controller's consumer of DEAD_LETTER_NOTICES and the " +
|
||||
"stream ASSIGNMENTS (novox/hq ADR 0582 clause 6): the recovery for a marker missed while the part was down, the " +
|
||||
"removal still started by the marker and nothing else. Refused unless the lease is handed to mesh-grants and the " +
|
||||
"removal has not run. Recorded as a hand act.",
|
||||
Input: schema(map[string]string{"why": "why, kept with the hand act", "ttl": "the time to live, e.g. 10s (1s to 24h)"}, "why", "ttl"),
|
||||
Run: func(a map[string]any) (any, error) {
|
||||
return p.RemovalDue(stdio.CallerOf(a), grants.Answer(a, "why"), grants.Answer(a, "ttl"))
|
||||
}},
|
||||
}
|
||||
}
|
||||
|
||||
// seatVerb is the seat mesh-grants' verb as mesh-sdk declares it.
|
||||
func seatVerb(name string) catalogue.Verb {
|
||||
seat, _ := catalogue.SeatNamed(catalogue.GrantsSeat)
|
||||
for _, v := range seat.Serves {
|
||||
if v.Name == name {
|
||||
return v
|
||||
}
|
||||
}
|
||||
return catalogue.Verb{Name: name}
|
||||
}
|
||||
|
||||
// serveTools serves the tools to the runtime until it stops.
|
||||
func serveTools() error {
|
||||
p := &grants.Part{Read: runtimeReader{}, Call: runtimeBus{}, Store: runtimeStore{}, Emit: runtimeBus{}}
|
||||
|
||||
@@ -398,8 +398,18 @@ func TestGrants(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatalf("mesh-sdk cannot decode the manifest: %v", err)
|
||||
}
|
||||
if m.Module != "mesh-grants" || len(m.Claims) != 0 {
|
||||
t.Errorf("module %q claims %v: the part claims no seat until mesh-sdk's seat list names mesh-grants", m.Module, m.Claims)
|
||||
// The part holds the seat mesh-grants for the lease write (ADR 0582 clause 1, issue 774, 774/S14), as mesh-sdk's seat
|
||||
// list declares it.
|
||||
if seat, ok := catalogue.SeatNamed(catalogue.GrantsSeat); m.Module != "mesh-grants" || len(m.Claims) != 1 || !ok ||
|
||||
catalogue.CanHold(m, seat) != nil {
|
||||
t.Errorf("module %q claims %v: the part holds mesh-grants alone (%v)", m.Module, m.Claims, catalogue.CanHold(m, seat))
|
||||
}
|
||||
var keyTTL bool
|
||||
for _, s := range m.State {
|
||||
keyTTL = keyTTL || (s.Name == grants.StateLease && s.KeyTTL)
|
||||
}
|
||||
if !keyTTL {
|
||||
t.Errorf("the state %s gives no time to live per key, which removal-due needs", grants.StateLease)
|
||||
}
|
||||
for _, row := range []struct {
|
||||
what string
|
||||
@@ -449,8 +459,9 @@ func TestGrants(t *testing.T) {
|
||||
if path != publishOnly && path != asksOnly && (strings.Contains(string(src), ".Publish(") || strings.Contains(string(src), "mesh.assignment")) {
|
||||
t.Errorf("%s publishes: only publish.go publishes memberships, and asking.go the operator's asks", path)
|
||||
}
|
||||
if path == asksOnly && (strings.Count(string(src), ".Publish(") != 2 || strings.Contains(string(src), "mesh.assignment")) {
|
||||
t.Errorf("asking.go publishes other than an ask, a cancel and a condition, through js.Publish twice")
|
||||
// And the seat mesh-grants' events (ADR 0582 clauses 3, 4, 6; issue 774).
|
||||
if path == asksOnly && (strings.Count(string(src), ".Publish(") != 3 || strings.Contains(string(src), "mesh.assignment")) {
|
||||
t.Errorf("asking.go publishes other than an ask, a cancel and a condition, and the seat's events, through js.Publish thrice")
|
||||
}
|
||||
if path == publishOnly && strings.Count(string(src), ".Publish(") != 1 {
|
||||
t.Errorf("publish.go publishes other than one membership")
|
||||
|
||||
@@ -20,7 +20,13 @@ type Watcher interface {
|
||||
// Followed are the states whose change composes again: placement, the indexer's manifests by module, the resolved
|
||||
// config values, the node agents' haves (a machine's first have, and the bus node's loaded list) and the assigner's
|
||||
// wants (the manifest each machine is wanted to run), and the vault's hashes of the bus accounts.
|
||||
var Followed = []string{ReadAssignments, ReadModules, ReadResolved, ReadHave, ReadWants, ReadAccounts, ReadPlacement, ReadLease, ReadTakeover}
|
||||
//
|
||||
// And its own state lease, whose key removal-due's expiry marker starts the day-later removal (ADR 0582 clause 6).
|
||||
var Followed = []string{ReadAssignments, ReadModules, ReadResolved, ReadHave, ReadWants, ReadAccounts, ReadPlacement, ReadLease, ReadTakeover,
|
||||
OwnLease}
|
||||
|
||||
// OwnLease is the part's own state lease as the follower watches it, `<module>.<state>`.
|
||||
const OwnLease = Module + "." + StateLease
|
||||
|
||||
// Follower composes again on every change it hears, never on a timer. It never stops for a refusal: a watch it cannot
|
||||
// open is said, composed without, and opened again at the next wake (any change it does hear, or a reconnect).
|
||||
@@ -121,14 +127,23 @@ func (f *Follower) heard(state, key string, value []byte) bool {
|
||||
}
|
||||
return false
|
||||
}
|
||||
if state == OwnLease {
|
||||
// The marker of removal-due (its value gone: expired, or deleted by a hand-back, which the removal tells apart).
|
||||
if key == KeyRemovalDue && value == nil && f.Part != nil {
|
||||
f.Part.RemovalDueHeard()
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
if state == ReadLease {
|
||||
// Renewed every few seconds by its holder: only the write that hands it away, or back, composes again (ADR 0561 §6).
|
||||
// Once handed away nobody renews it, so any change is heard: it may be the lease altered (ADR 0582 clause 4).
|
||||
on, _ := buscompose.LeaseHandedAway(value, len(value) > 0, nil)
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
was, seen := f.leaseOn, f.leaseSeen
|
||||
f.leaseOn, f.leaseSeen = on, true
|
||||
return !seen || was != on
|
||||
return !seen || was != on || on
|
||||
}
|
||||
if state != ReadHave {
|
||||
return true
|
||||
@@ -169,7 +184,8 @@ func (f *Follower) open() {
|
||||
}
|
||||
if f.Hear != nil && !f.heard_ {
|
||||
err := f.Hear(func(event string, body []byte) {
|
||||
if IsAccountIssued(event) {
|
||||
// An account the vault minted, or a bucket's expiry set (the lease's, altered: ADR 0582 clause 4).
|
||||
if IsAccountIssued(event) || IsBucketExpirySet(event) {
|
||||
f.Wake()
|
||||
return
|
||||
}
|
||||
|
||||
@@ -52,7 +52,7 @@ const (
|
||||
)
|
||||
|
||||
// Invokes is every tool the part calls.
|
||||
var Invokes = []string{CallUsers, CallUserCan, CallStreams, CallRootFree, CallRecordWarrant}
|
||||
var Invokes = append([]string{CallUsers, CallUserCan, CallStreams, CallRootFree, CallRecordWarrant}, LeaseCalls...)
|
||||
|
||||
// Reader reads another part's state, named `<module>.<state>`. A refusal (a read not granted yet, a bucket not there) is
|
||||
// an error the part names as a gap and composes past: it never stops for one.
|
||||
|
||||
@@ -0,0 +1,945 @@
|
||||
package grants
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/handacts"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
)
|
||||
|
||||
// **The lease write, a verb of the seat mesh-grants** (novox/hq ADR 0582, issue 774). `lease-write` takes a why and is
|
||||
// refused, naming the check, until every precondition of clause 2 reads green; otherwise the operator is asked once, on
|
||||
// a channel that proves who answered, with the checks shown and their digest bound to the yes (clause 1). On the
|
||||
// warrant the checks are read again, and then, each step read back before the next (clause 3): the controller's service
|
||||
// is stopped on its node and read back stopped; the bus's user list is composed without the controller's write on the
|
||||
// lease (buscompose.Records.LeaseCut) and waited on until the bus node's have says it loaded it; the lease bucket's
|
||||
// expiry is removed through the bus module's tool; the key `holder` names mesh-grants with the operator, the why, the
|
||||
// ask and the time; `lease-handed` is said, the act recorded through mesh-hand-acts.warranted, and `removal-due` is
|
||||
// written in the state `lease` with a day to live. A lease altered after it is said as `lease-altered`, once, and
|
||||
// nothing is written again until the operator asks (clause 4). `lease-hand-back` reverses it for a day (clause 5). The
|
||||
// expiry marker of `removal-due`, and nothing else, starts the day-later removal of the controller's consumer of
|
||||
// DEAD_LETTER_NOTICES and the stream ASSIGNMENTS (clause 6). Every step waits on an event (a warrant, the bus node's
|
||||
// have, a marker), never on a timer.
|
||||
|
||||
// The verbs, events and states of the lease write.
|
||||
const (
|
||||
ActLeaseWrite = catalogue.GrantsVerbLeaseWrite
|
||||
ActLeaseHandBack = catalogue.GrantsVerbLeaseHandBack
|
||||
EventLeaseHanded = catalogue.GrantsEventLeaseHanded
|
||||
// EventLeaseAltered and EventRemovalFailed are the seat's other two events.
|
||||
EventLeaseAltered = catalogue.GrantsEventLeaseAltered
|
||||
EventRemovalFailed = catalogue.GrantsEventRemovalFailed
|
||||
// StateLease is the part's state with a time to live per key, holding KeyRemovalDue alone.
|
||||
StateLease = "lease"
|
||||
KeyRemovalDue = "removal-due"
|
||||
// KeyLease is the lease write's record in StateTakeover: the cut, the write, the removal and every act.
|
||||
KeyLease = "lease"
|
||||
// ToolRemovalDue is the tool that writes removal-due again with a short time to live, the recovery for a marker
|
||||
// missed while the part was down.
|
||||
ToolRemovalDue = "grants_removal_due"
|
||||
)
|
||||
|
||||
// RemovalAfter is how long after the write the controller's consumer and stream are removed (ADR 0582 clause 6).
|
||||
const RemovalAfter = 24 * time.Hour
|
||||
|
||||
// The calls the lease write makes, as the manifest's invokes name them.
|
||||
const (
|
||||
// CallServiceStop, CallServiceStart and CallServiceStatus are the node seat's verbs, asked of the controller's node.
|
||||
CallServiceStop = "seat:" + catalogue.ServiceManagerSeat + ".stop"
|
||||
CallServiceStart = "seat:" + catalogue.ServiceManagerSeat + ".start"
|
||||
CallServiceStatus = "seat:" + catalogue.ServiceManagerSeat + ".status"
|
||||
// CallBucketExpiry sets and restores a bucket's expiry, recording the act itself; CallConsumerRemove and
|
||||
// CallStreamRemove are the removal's two tools.
|
||||
CallBucketExpiry = "nats.nats_bucket_expiry"
|
||||
CallConsumerRemove = "nats.nats_consumer_remove"
|
||||
CallStreamRemove = "nats.nats_stream_remove"
|
||||
// CallRecordHandAct records the removals as hand acts (ADR 0582 clause 6, ADR 0565 §11).
|
||||
CallRecordHandAct = "seat:" + handacts.Seat + ".hand-act"
|
||||
// The checks' calls (clause 2).
|
||||
CallNodeTrusts = "mesh-delivery-view.node_trusts"
|
||||
CallConfigCompare = "seat:mesh-config.compare"
|
||||
CallUpgrade = "seat:" + catalogue.WantsSeat + ".upgrade"
|
||||
CallVaultStatus = "mesh-vault.vault_status"
|
||||
CallIssue = "seat:issue-tracker.get"
|
||||
)
|
||||
|
||||
// LeaseCalls are every call the lease write makes.
|
||||
var LeaseCalls = []string{CallServiceStop, CallServiceStart, CallServiceStatus, CallBucketExpiry, CallConsumerRemove,
|
||||
CallStreamRemove, CallRecordHandAct, CallNodeTrusts, CallConfigCompare, CallUpgrade, CallVaultStatus, CallIssue}
|
||||
|
||||
// ControllerUnit is the controller's service on its node.
|
||||
const ControllerUnit = Controller + ".service"
|
||||
|
||||
// LeaseExpiry is the lease bucket's expiry a hand-back restores (buscompose.LeaseTTL).
|
||||
var LeaseExpiry = buscompose.LeaseTTL
|
||||
|
||||
// RemovedTools are the bus module's tools the lease write and its removal need: the check reads them in its manifest.
|
||||
var RemovedTools = []string{"nats_bucket_expiry", "nats_consumer_remove", "nats_stream_remove"}
|
||||
|
||||
// EarlierSteps are the issues of design 47's earlier steps the write waits on (ADR 0582 clause 2: L4 to L7 delivered).
|
||||
var EarlierSteps = []uint64{1952, 1953, 1954, 1955}
|
||||
|
||||
// LeaseBus is the process's hand on the controller's lease: the key `holder` put and deleted, and the bucket's expiry
|
||||
// read. Nil in the tools, which ask the process.
|
||||
type LeaseBus interface {
|
||||
PutHolder(value []byte) error
|
||||
DeleteHolder() error
|
||||
Expiry() (time.Duration, error)
|
||||
}
|
||||
|
||||
// TTLStore is a Store that writes a key with its own time to live (the state lease).
|
||||
type TTLStore interface {
|
||||
PutTTL(state, key string, value []byte, ttl time.Duration) (uint64, error)
|
||||
}
|
||||
|
||||
// SeatSayer says one of the seat's events, `mesh.seat.mesh-grants.event.<event>`.
|
||||
type SeatSayer func(event string, body any) error
|
||||
|
||||
// Check is one precondition as read: its name, whether it reads green, and what it says.
|
||||
type Check struct {
|
||||
Name string `json:"name"`
|
||||
Green bool `json:"green"`
|
||||
Says string `json:"says"`
|
||||
}
|
||||
|
||||
// checksDigest is the digest of the checks as shown: what the yes binds.
|
||||
func checksDigest(cs []Check) string {
|
||||
raw, _ := json.Marshal(cs)
|
||||
sum := sha256.Sum256(raw)
|
||||
return "sha256:" + hex.EncodeToString(sum[:])
|
||||
}
|
||||
|
||||
// LeaseAct is one lease-write or lease-hand-back: asked, answered and each step it reached.
|
||||
type LeaseAct struct {
|
||||
Act string `json:"act"`
|
||||
By string `json:"by"`
|
||||
Why string `json:"why"`
|
||||
At time.Time `json:"at"`
|
||||
Ask string `json:"ask"`
|
||||
// Checks are the preconditions as the ask showed them, and Digest their digest the yes binds.
|
||||
Checks []Check `json:"checks,omitempty"`
|
||||
Digest string `json:"digest,omitempty"`
|
||||
Raised bool `json:"raised,omitempty"`
|
||||
// Step is where it stands: asked; then on the warrant stopped, cut, loaded, expiry, written, done for a write and
|
||||
// restored, uncut, done for a hand-back; or declined, expired, refused, failed.
|
||||
Step string `json:"step"`
|
||||
Said string `json:"said,omitempty"`
|
||||
// List is the digest of the list composed at the cut (or without it, for a hand-back), awaited loaded.
|
||||
List string `json:"list,omitempty"`
|
||||
Done time.Time `json:"done,omitzero"`
|
||||
}
|
||||
|
||||
// The steps an act reaches.
|
||||
const (
|
||||
StepAsked = "asked"
|
||||
StepStopped = "stopped"
|
||||
StepCut = "cut"
|
||||
StepWritten = "written"
|
||||
StepRestored = "restored"
|
||||
StepUncut = "uncut"
|
||||
StepDone = "done"
|
||||
StepDeclined = "declined"
|
||||
StepExpired = "expired"
|
||||
StepRefused = "refused"
|
||||
StepFailed = "failed"
|
||||
)
|
||||
|
||||
// LeaseRecord is the key `lease` of StateTakeover.
|
||||
type LeaseRecord struct {
|
||||
V int `json:"v"`
|
||||
// Cut composes the list with buscompose.Records.LeaseCut: from the write's second step until a hand-back's.
|
||||
Cut bool `json:"cut"`
|
||||
// Node is the controller's node, from the lease's holder at the write.
|
||||
Node string `json:"node,omitempty"`
|
||||
// Handed is when the key holder was written, zero when not (or handed back); Ask the write's ask.
|
||||
Handed time.Time `json:"handed,omitzero"`
|
||||
Ask string `json:"ask,omitempty"`
|
||||
// Removed is when the day-later removal was done; zero until then.
|
||||
Removed time.Time `json:"removed,omitzero"`
|
||||
// Altered is what lease-altered said, empty while the lease reads as written.
|
||||
Altered string `json:"altered,omitempty"`
|
||||
// Open is the act asked or under way.
|
||||
Open *LeaseAct `json:"open,omitempty"`
|
||||
// Acts are the acts ended, newest last.
|
||||
Acts []LeaseAct `json:"acts"`
|
||||
}
|
||||
|
||||
func (p *Part) readLease() (LeaseRecord, error) {
|
||||
r := LeaseRecord{V: 1, Acts: []LeaseAct{}}
|
||||
raw, _, found, err := p.Store.Get(StateTakeover, KeyLease)
|
||||
if err != nil {
|
||||
return r, fmt.Errorf("the lease write's record cannot be read: %w", err)
|
||||
}
|
||||
if found {
|
||||
if err := json.Unmarshal(raw, &r); err != nil {
|
||||
return r, fmt.Errorf("the lease write's record does not read: %w", err)
|
||||
}
|
||||
}
|
||||
return r, nil
|
||||
}
|
||||
|
||||
func (p *Part) writeLease(r LeaseRecord) error {
|
||||
if len(r.Acts) > keptActs {
|
||||
r.Acts = r.Acts[len(r.Acts)-keptActs:]
|
||||
}
|
||||
if r.Acts == nil {
|
||||
r.Acts = []LeaseAct{}
|
||||
}
|
||||
data, _ := json.Marshal(r)
|
||||
if _, err := p.Store.Put(StateTakeover, KeyLease, data); err != nil {
|
||||
return fmt.Errorf("the lease write's record cannot be written: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// leaseCut says whether the list is composed with the cut.
|
||||
func (p *Part) leaseCut() bool {
|
||||
r, err := p.readLease()
|
||||
return err == nil && r.Cut
|
||||
}
|
||||
|
||||
// leaseHanded says whether the key holder names mesh-grants by the part's own write.
|
||||
func (p *Part) leaseHanded() bool {
|
||||
r, err := p.readLease()
|
||||
return err == nil && !r.Handed.IsZero()
|
||||
}
|
||||
|
||||
// end closes the open act with its step and why.
|
||||
func (r *LeaseRecord) end(step, said string, at time.Time) {
|
||||
if r.Open == nil {
|
||||
return
|
||||
}
|
||||
a := *r.Open
|
||||
a.Step, a.Said, a.Done = step, said, at
|
||||
r.Acts = append(r.Acts, a)
|
||||
r.Open = nil
|
||||
}
|
||||
|
||||
// LeaseCheckNames are the checks LeaseChecks reads, in its order.
|
||||
var LeaseCheckNames = []string{"lease-check", "earlier-steps", "node-trusts", "config-compare", "upgrade-compare",
|
||||
"keeper-take-over", "vault-minting", "bus-tools"}
|
||||
|
||||
// LeaseChecks reads every precondition of the write (ADR 0582 clause 2), each named, in one order.
|
||||
func (p *Part) LeaseChecks() []Check {
|
||||
var out []Check
|
||||
add := func(name string, green bool, says string) {
|
||||
out = append(out, Check{Name: name, Green: green, Says: says})
|
||||
}
|
||||
// The lease check at the write, with the three kept to it alone (ADR 0571 §13).
|
||||
if raw, err := p.LeaseCheck(); err != nil {
|
||||
add("lease-check", false, "the lease check could not run: "+err.Error())
|
||||
} else if a, ok := raw.(LeaseAnswer); !ok {
|
||||
add("lease-check", false, "the lease check answered nothing it reads")
|
||||
} else {
|
||||
add("lease-check", a.Check.Passes, strings.Join(a.Check.Reasons, "; "))
|
||||
}
|
||||
// Design 47's earlier steps delivered.
|
||||
var open []string
|
||||
for _, n := range EarlierSteps {
|
||||
raw, err := p.Call.Call(CallIssue, map[string]any{"number": n, "brief": true})
|
||||
var is struct {
|
||||
Status string `json:"status"`
|
||||
}
|
||||
switch {
|
||||
case err != nil:
|
||||
open = append(open, fmt.Sprintf("issue %d cannot be read (%v)", n, err))
|
||||
case json.Unmarshal(unwrap(raw), &is) != nil || is.Status != "resolved":
|
||||
open = append(open, fmt.Sprintf("issue %d is %q, not resolved", n, is.Status))
|
||||
}
|
||||
}
|
||||
add("earlier-steps", len(open) == 0, said(open, "issues 1952 to 1955 are resolved"))
|
||||
// Every node trusts the vault's key.
|
||||
if raw, err := p.Call.Call(CallNodeTrusts, map[string]any{}); err != nil {
|
||||
add("node-trusts", false, "node_trusts cannot be read: "+err.Error())
|
||||
} else {
|
||||
var t struct {
|
||||
Every bool `json:"every"`
|
||||
NotYet []string `json:"not_yet"`
|
||||
}
|
||||
_ = json.Unmarshal(unwrap(raw), &t)
|
||||
add("node-trusts", t.Every, said(prefixed("not yet: ", t.NotYet), "every node's record trusts the vault's key"))
|
||||
}
|
||||
// mesh-config's compare empty and its take-over done.
|
||||
if raw, err := p.Call.Call(CallConfigCompare, map[string]any{}); err != nil {
|
||||
add("config-compare", false, "mesh-config's compare cannot be read: "+err.Error())
|
||||
} else {
|
||||
var c struct {
|
||||
Source string `json:"source"`
|
||||
Count int `json:"count"`
|
||||
}
|
||||
_ = json.Unmarshal(unwrap(raw), &c)
|
||||
add("config-compare", c.Count == 0 && c.Source == "indexer",
|
||||
fmt.Sprintf("mesh-config resolves from %q with %d differences", c.Source, c.Count))
|
||||
}
|
||||
// The upgrade compare empty.
|
||||
if raw, err := p.Call.Call(CallUpgrade, map[string]any{"compare": "true"}); err != nil {
|
||||
add("upgrade-compare", false, "mesh-wants' upgrade compare cannot be read: "+err.Error())
|
||||
} else {
|
||||
var u struct {
|
||||
Empty bool `json:"empty"`
|
||||
Splits []json.RawMessage `json:"splits"`
|
||||
}
|
||||
_ = json.Unmarshal(unwrap(raw), &u)
|
||||
add("upgrade-compare", u.Empty, fmt.Sprintf("%d policy splits", len(u.Splits)))
|
||||
}
|
||||
// The condition keeper's take-over done.
|
||||
raw, found, err := p.Read.Get(ReadTakeover, health.TakeoverKey)
|
||||
t, terr := health.DecodeTakeover(raw, found)
|
||||
switch {
|
||||
case err != nil:
|
||||
add("keeper-take-over", false, "the keeper's take-over cannot be read: "+err.Error())
|
||||
case terr != nil:
|
||||
add("keeper-take-over", false, "the keeper's take-over does not read: "+terr.Error())
|
||||
default:
|
||||
add("keeper-take-over", t.Follow != health.FollowController, "conditions follow "+t.Follow)
|
||||
}
|
||||
// The vault minting with nothing held back.
|
||||
if raw, err := p.Call.Call(CallVaultStatus, map[string]any{}); err != nil {
|
||||
add("vault-minting", false, "vault_status cannot be read: "+err.Error())
|
||||
} else {
|
||||
var v struct {
|
||||
Ready bool `json:"ready"`
|
||||
Minting struct {
|
||||
On bool `json:"on"`
|
||||
HeldBack map[string]string `json:"held_back"`
|
||||
} `json:"minting"`
|
||||
}
|
||||
_ = json.Unmarshal(unwrap(raw), &v)
|
||||
add("vault-minting", v.Ready && v.Minting.On && len(v.Minting.HeldBack) == 0,
|
||||
fmt.Sprintf("ready %v, minting %v, %d held back", v.Ready, v.Minting.On, len(v.Minting.HeldBack)))
|
||||
}
|
||||
// The bus module's tools for the expiry and the removal live: its manifest at the indexer's trunk lists them.
|
||||
in := ReadInputs(p.Read, &p.cache)
|
||||
var missing []string
|
||||
for _, tool := range RemovedTools {
|
||||
if !slices.Contains(in.Manifests[BusModule].Tools, tool) {
|
||||
missing = append(missing, tool)
|
||||
}
|
||||
}
|
||||
add("bus-tools", len(missing) == 0, said(prefixed("the bus module lists no ", missing), "the bus module lists its three tools"))
|
||||
return out
|
||||
}
|
||||
|
||||
func said(problems []string, ok string) string {
|
||||
if len(problems) == 0 {
|
||||
return ok
|
||||
}
|
||||
return strings.Join(problems, "; ")
|
||||
}
|
||||
|
||||
func prefixed(prefix string, items []string) []string {
|
||||
if len(items) == 0 {
|
||||
return nil
|
||||
}
|
||||
return []string{prefix + strings.Join(items, ", ")}
|
||||
}
|
||||
|
||||
// notGreen is every check not green, `<name>: <says>`.
|
||||
func notGreen(cs []Check) []string {
|
||||
var out []string
|
||||
for _, c := range cs {
|
||||
if !c.Green {
|
||||
out = append(out, c.Name+": "+c.Says)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// changedChecks names each check whose reading differs from the one shown.
|
||||
func changedChecks(shown, now []Check) []string {
|
||||
was := map[string]Check{}
|
||||
for _, c := range shown {
|
||||
was[c.Name] = c
|
||||
}
|
||||
var out []string
|
||||
for _, c := range now {
|
||||
if w, ok := was[c.Name]; !ok || w != c {
|
||||
out = append(out, c.Name+" (now: "+c.Says+")")
|
||||
}
|
||||
delete(was, c.Name)
|
||||
}
|
||||
for name := range was {
|
||||
out = append(out, name+" (no longer read)")
|
||||
}
|
||||
slices.Sort(out)
|
||||
return out
|
||||
}
|
||||
|
||||
// LeaseWrite is the verb lease-write: refused, naming each check not green, or asked of the operator by the part's process.
|
||||
func (p *Part) LeaseWrite(caller, why string) (any, error) {
|
||||
return p.leaseRequest(ActLeaseWrite, caller, why)
|
||||
}
|
||||
|
||||
// LeaseHandBack is the verb lease-hand-back.
|
||||
func (p *Part) LeaseHandBack(caller, why string) (any, error) {
|
||||
return p.leaseRequest(ActLeaseHandBack, caller, why)
|
||||
}
|
||||
|
||||
func (p *Part) leaseRequest(act, caller, why string) (any, error) {
|
||||
if strings.TrimSpace(why) == "" {
|
||||
return nil, fmt.Errorf("%s needs a why, shown to the operator and kept", act)
|
||||
}
|
||||
r, err := p.readLease()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if r.Open != nil {
|
||||
return nil, fmt.Errorf("%s is already under way: the ask %s stands at %s", r.Open.Act, r.Open.Ask, r.Open.Step)
|
||||
}
|
||||
var checks []Check
|
||||
switch act {
|
||||
case ActLeaseWrite:
|
||||
if !r.Handed.IsZero() && r.Altered == "" {
|
||||
return nil, errors.New("the lease is handed to mesh-grants already: nothing to write")
|
||||
}
|
||||
checks = p.LeaseChecks()
|
||||
if bad := notGreen(checks); len(bad) > 0 {
|
||||
return nil, fmt.Errorf("lease-write is refused until every check reads green (ADR 0582 clause 2): %s", strings.Join(bad, "; "))
|
||||
}
|
||||
case ActLeaseHandBack:
|
||||
if refused := handBackRefused(r); refused != "" {
|
||||
return nil, errors.New(refused)
|
||||
}
|
||||
}
|
||||
a := LeaseAct{Act: act, By: caller, Why: why, At: p.now(), Checks: checks, Digest: checksDigest(checks), Step: StepAsked}
|
||||
a.Ask = act + "-" + p.now().Format("20060102t150405") + "-" + p.nonce()
|
||||
r.Open = &a
|
||||
if err := p.writeLease(r); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
_, _ = p.AskCompose(caller)
|
||||
return map[string]any{"asked": a.Ask, "checks": checks, "says": "the part's process asks the operator on their channel; " +
|
||||
"only the operator's yes runs it, after the checks are read again. grants_lease_check shows the lease, and " +
|
||||
"the record is the key lease of the state takeover"}, nil
|
||||
}
|
||||
|
||||
// handBackRefused is why a hand-back cannot be asked, or empty.
|
||||
func handBackRefused(r LeaseRecord) string {
|
||||
switch {
|
||||
case !r.Removed.IsZero():
|
||||
return "lease-hand-back is refused: the day-later removal ran, so the controller would start without its " +
|
||||
"consumer of DEAD_LETTER_NOTICES and the stream ASSIGNMENTS (ADR 0582 clause 5)"
|
||||
case r.Handed.IsZero() && !r.Cut:
|
||||
return "lease-hand-back is refused: the lease is not handed to mesh-grants, so there is nothing to hand back"
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// EventBucketExpirySet is the bus module's event of a bucket's expiry set, which the part consumes: a lease bucket's
|
||||
// expiry changed is heard, never polled.
|
||||
const EventBucketExpirySet = BusModule + ".bucket-expiry-set"
|
||||
|
||||
// IsBucketExpirySet says an event heard is the bus module's bucket-expiry-set.
|
||||
func IsBucketExpirySet(event string) bool {
|
||||
return event == EventBucketExpirySet || event == "mesh.mod."+BusModule+".event.bucket-expiry-set"
|
||||
}
|
||||
|
||||
// isLeaseAsk says an ask is the lease write's.
|
||||
func isLeaseAsk(id string) bool {
|
||||
return strings.HasPrefix(id, ActLeaseWrite+"-") || strings.HasPrefix(id, ActLeaseHandBack+"-")
|
||||
}
|
||||
|
||||
// leaseAskOf is the ask of an act, in the words the operator reads: what it does, the checks it showed, and the one
|
||||
// option whose yes binds the act, the holder, the ask and the checks' digest (ADR 0582 clause 1).
|
||||
func (p *Part) leaseAskOf(a LeaseAct) (asks.Ask, error) {
|
||||
act := asks.Act{"verb": a.Act, "module": Module, "ask": a.Ask, "holder": Module, "checks": a.Digest}
|
||||
bind, err := asks.ActDigest(act)
|
||||
if err != nil {
|
||||
return asks.Ask{}, err
|
||||
}
|
||||
var lines []string
|
||||
for _, c := range a.Checks {
|
||||
lines = append(lines, "✓ "+c.Name+": "+c.Says)
|
||||
}
|
||||
headline, does, stays := "Hand the controller's lease to mesh-grants?",
|
||||
"The controller is stopped and the lease is handed to mesh-grants for good; hand-back stays open for a day.",
|
||||
"The controller keeps its lease."
|
||||
expl := "Every check of the lease write reads green. Your yes stops the controller, takes from it the right to write " +
|
||||
"its lease, removes the lease's expiry and names mesh-grants as its holder. A controller started later fails on the lease."
|
||||
if a.Act == ActLeaseHandBack {
|
||||
headline, does, stays = "Give the controller its lease back?",
|
||||
"The lease's expiry is restored, mesh-grants lets go of it, the controller gets its rights back and is started.",
|
||||
"The lease stays with mesh-grants."
|
||||
expl = "The lease was handed to mesh-grants less than a day ago. Your yes gives it back and starts the controller."
|
||||
}
|
||||
return asks.Ask{ID: a.Ask, Headline: headline, Explanation: expl, Who: asks.Operator,
|
||||
Options: []asks.Option{{ID: "yes", Label: "Yes", Does: does, Level: asks.Approve, Binds: bind},
|
||||
{ID: "no", Label: "No, leave it", Does: stays, Level: asks.Acknowledge}},
|
||||
Expires: a.At.Add(asks.ApproveLasts - time.Minute), OnExpiry: stays + " Nothing changes.", About: "mesh-grants-" + a.Act,
|
||||
Details: "Asked by " + a.By + ": " + a.Why + "\n" + strings.Join(lines, "\n")}, nil
|
||||
}
|
||||
|
||||
// raiseLease asks the operator the open act once; closes one past its expiry; catches up a warrant the process missed.
|
||||
func (p *Part) raiseLease() error {
|
||||
if p.Asker == nil {
|
||||
return nil
|
||||
}
|
||||
r, err := p.readLease()
|
||||
if err != nil || r.Open == nil || r.Open.Step != StepAsked {
|
||||
return err
|
||||
}
|
||||
a := r.Open
|
||||
if !a.Raised {
|
||||
ask, err := p.leaseAskOf(*a)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := p.Asker.Ask(ask); err != nil {
|
||||
if p.Forget != nil {
|
||||
_ = p.Forget(a.Ask)
|
||||
}
|
||||
_ = p.Asker.Cancel(a.Ask)
|
||||
r.end(StepRefused, "the operator could not be asked: "+err.Error(), p.now())
|
||||
_ = p.writeLease(r)
|
||||
return fmt.Errorf("the operator could not be asked the %s: %v", a.Act, err)
|
||||
}
|
||||
a.Raised = true
|
||||
return p.writeLease(r)
|
||||
}
|
||||
_, w, err := p.Asker.Where(a.Ask)
|
||||
if err != nil {
|
||||
return fmt.Errorf("the router's record of the ask %s cannot be read: %v", a.Ask, err)
|
||||
}
|
||||
if w != nil {
|
||||
h, herr := p.Asker.Take(*w, p.Warranted)
|
||||
if h == "" && herr != nil {
|
||||
return fmt.Errorf("the operator's answer to %s could not be taken yet: %v", a.Ask, herr)
|
||||
}
|
||||
p.Said(h, *w, herr)
|
||||
return nil
|
||||
}
|
||||
if p.now().After(a.At.Add(asks.ApproveLasts + askMargin)) {
|
||||
r.end(StepExpired, "the ask expired: nothing was written", p.now())
|
||||
return p.writeLease(r)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// leaseWarranted is the operator's answer to a lease act: a no or a refusal writes nothing; a yes runs the act.
|
||||
func (p *Part) leaseWarranted(o asks.Option, w asks.Warrant) error {
|
||||
r, err := p.readLease()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if r.Open == nil || r.Open.Ask != w.Ask || r.Open.Step != StepAsked {
|
||||
return fmt.Errorf("the warrant answers the ask %s, which is not the lease act asked: nothing is written", w.Ask)
|
||||
}
|
||||
if o.ID != "yes" {
|
||||
r.end(StepDeclined, "the operator said no: nothing was written", p.now())
|
||||
return p.writeLease(r)
|
||||
}
|
||||
a := r.Open
|
||||
if a.Act == ActLeaseWrite {
|
||||
// Read again on the warrant (ADR 0582 clauses 1, 2): the yes binds the checks shown, so any check that reads
|
||||
// otherwise now (one no longer green among them) refuses it, named.
|
||||
if now := p.LeaseChecks(); checksDigest(now) != a.Digest {
|
||||
r.end(StepRefused, "refused on the warrant, the checks changed since the ask: "+strings.Join(changedChecks(a.Checks, now), "; "), p.now())
|
||||
p.sayLeaseRefused(a.Act, r.Acts[len(r.Acts)-1].Said)
|
||||
return p.writeLease(r)
|
||||
}
|
||||
} else if refused := handBackRefused(r); refused != "" {
|
||||
r.end(StepRefused, refused, p.now())
|
||||
return p.writeLease(r)
|
||||
}
|
||||
a.Step = "warranted"
|
||||
if err := p.writeLease(r); err != nil {
|
||||
return err
|
||||
}
|
||||
return p.advanceLease()
|
||||
}
|
||||
|
||||
// sayLeaseRefused tells the operator a yes was refused, as the take-over's refusal is said.
|
||||
func (p *Part) sayLeaseRefused(act, why string) {
|
||||
if p.Say == nil {
|
||||
return
|
||||
}
|
||||
_ = p.Say(asks.Condition{Key: "mesh-grants." + act + "-refused", Severity: asks.SeverityWarning,
|
||||
Subject: asks.Scope{Scope: "module", ID: Module}, State: asks.StateRaised, Summary: act + " refused: " + why,
|
||||
Headline: "The lease write was refused", Needs: "a look at why",
|
||||
Explanation: asks.NeedsYou + "a look at why. You said yes and it was refused: " + why, Resolved: "the act is asked again and goes through"})
|
||||
}
|
||||
|
||||
// advanceLease takes the open act as far as it can now: each step read back before the next, and a step that waits (the
|
||||
// list loaded) left for the pass that hears the bus node's have. A step that fails ends the act there, saying why, and
|
||||
// writes nothing more.
|
||||
func (p *Part) advanceLease() error {
|
||||
r, err := p.readLease()
|
||||
if err != nil || r.Open == nil || r.Open.Step == StepAsked {
|
||||
return err
|
||||
}
|
||||
a := r.Open
|
||||
fail := func(why string) error {
|
||||
r.end(StepFailed, why, p.now())
|
||||
p.sayLeaseRefused(a.Act, why)
|
||||
return p.writeLease(r)
|
||||
}
|
||||
if a.Act == ActLeaseHandBack {
|
||||
return p.advanceHandBack(&r, fail)
|
||||
}
|
||||
switch a.Step {
|
||||
case "warranted":
|
||||
// 1. Stop the controller's service on its node, read back stopped.
|
||||
node, why := p.controllerNode()
|
||||
if node == "" {
|
||||
return fail(why)
|
||||
}
|
||||
if stopped, why := p.service(CallServiceStop, node); !stopped {
|
||||
return fail("the controller's service on " + node + " did not read back stopped (" + why + "): nothing more is written")
|
||||
}
|
||||
r.Node, a.Step = node, StepStopped
|
||||
fallthrough
|
||||
case StepStopped:
|
||||
// 2. Compose the list without the controller's write on the lease.
|
||||
r.Cut, a.Step = true, StepCut
|
||||
if err := p.writeLease(r); err != nil {
|
||||
return err
|
||||
}
|
||||
res := p.composeQuiet()
|
||||
g, _, _ := p.Sealed()
|
||||
if res.Sealed == nil || g.Digest == "" {
|
||||
return fail("the list composed with the cut could not be written sealed: " + strings.Join(res.Errors, "; "))
|
||||
}
|
||||
if r, err = p.readLease(); err != nil || r.Open == nil {
|
||||
return err
|
||||
}
|
||||
a = r.Open
|
||||
a.List = g.Digest
|
||||
if err := p.writeLease(r); err != nil {
|
||||
return err
|
||||
}
|
||||
fallthrough
|
||||
case StepCut:
|
||||
// 3. Wait until the bus node's have says it loaded that list.
|
||||
if !p.loaded(a.List) {
|
||||
return nil
|
||||
}
|
||||
// 4. Remove the lease bucket's expiry.
|
||||
if ok, why := p.setExpiry("none", *a); !ok {
|
||||
return fail("the lease's expiry was not removed (" + why + "): the holder is not written")
|
||||
}
|
||||
// 5. Create the key holder naming mesh-grants.
|
||||
h := buscompose.LeaseHolder{HandedTo: Module, HandedBy: "operator", Why: a.Why, Ask: a.Ask, At: p.now(), Host: r.Node}
|
||||
raw, _ := json.Marshal(h)
|
||||
if p.Lease == nil {
|
||||
return fail("the part holds no hand on the lease here")
|
||||
}
|
||||
if err := p.Lease.PutHolder(raw); err != nil {
|
||||
return fail("the key holder could not be written: " + err.Error())
|
||||
}
|
||||
r.Handed, r.Ask, r.Altered, a.Step = p.now(), a.Ask, "", StepWritten
|
||||
if err := p.writeLease(r); err != nil {
|
||||
return err
|
||||
}
|
||||
// 6. Say lease-handed, record the act, and write removal-due with a day to live.
|
||||
p.saySeat(EventLeaseHanded, map[string]any{"holder": Module, "ask": a.Ask, "why": a.Why, "node": r.Node, "at": r.Handed})
|
||||
p.recordWarrant(a.Ask)
|
||||
if err := p.putRemovalDue(RemovalAfter); err != nil {
|
||||
r.end(StepDone, "the lease is handed; removal-due could not be written ("+err.Error()+"): grants_removal_due writes it", p.now())
|
||||
} else {
|
||||
r.end(StepDone, "the lease is handed to mesh-grants; the removal is due in a day", p.now())
|
||||
}
|
||||
return p.writeLease(r)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// advanceHandBack restores the expiry, deletes holder and removal-due, composes without the cut, and once that list is
|
||||
// loaded starts the controller (ADR 0582 clause 5).
|
||||
func (p *Part) advanceHandBack(r *LeaseRecord, fail func(string) error) error {
|
||||
a := r.Open
|
||||
switch a.Step {
|
||||
case "warranted":
|
||||
if ok, why := p.setExpiry(LeaseExpiry.String(), *a); !ok {
|
||||
return fail("the lease's expiry was not restored (" + why + "): the lease stays with mesh-grants")
|
||||
}
|
||||
if p.Lease == nil {
|
||||
return fail("the part holds no hand on the lease here")
|
||||
}
|
||||
// The record first: the delete of removal-due is then heard as no removal (Handed is zero).
|
||||
r.Handed, a.Step = time.Time{}, StepRestored
|
||||
if err := p.writeLease(*r); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := p.Lease.DeleteHolder(); err != nil {
|
||||
return fail("the key holder could not be deleted: " + err.Error())
|
||||
}
|
||||
if err := p.Store.Delete(StateLease, KeyRemovalDue); err != nil {
|
||||
return fail("removal-due could not be deleted: " + err.Error())
|
||||
}
|
||||
r.Cut, a.Step = false, StepUncut
|
||||
if err := p.writeLease(*r); err != nil {
|
||||
return err
|
||||
}
|
||||
p.composeQuiet()
|
||||
g, _, _ := p.Sealed()
|
||||
fresh, err := p.readLease()
|
||||
if err != nil || fresh.Open == nil {
|
||||
return err
|
||||
}
|
||||
*r = fresh
|
||||
a = r.Open
|
||||
a.List = g.Digest
|
||||
if err := p.writeLease(*r); err != nil {
|
||||
return err
|
||||
}
|
||||
fallthrough
|
||||
case StepUncut:
|
||||
if !p.loaded(a.List) {
|
||||
return nil
|
||||
}
|
||||
if started, why := p.service(CallServiceStart, r.Node); !started {
|
||||
return fail("the controller's service on " + r.Node + " did not read back started (" + why + ")")
|
||||
}
|
||||
p.recordWarrant(a.Ask)
|
||||
r.end(StepDone, "the lease is given back; the controller is started", p.now())
|
||||
return p.writeLease(*r)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// controllerNode is the node the controller holds its lease on, from the lease's holder.
|
||||
func (p *Part) controllerNode() (string, string) {
|
||||
raw, found, err := p.Read.Get(ReadLease, LeaseKey)
|
||||
if err != nil || !found {
|
||||
return "", "the lease cannot be read, so the controller's node is not known: nothing is stopped"
|
||||
}
|
||||
var h buscompose.LeaseHolder
|
||||
if json.Unmarshal(raw, &h) != nil || h.Host == "" || h.HandedTo != "" {
|
||||
return "", "the lease names no controller's host, so the controller's node is not known: nothing is stopped"
|
||||
}
|
||||
return h.Host, ""
|
||||
}
|
||||
|
||||
// service asks the node seat's verb of the controller's unit and reads it back with status: stopped for a stop, started
|
||||
// for a start.
|
||||
func (p *Part) service(verb, node string) (bool, string) {
|
||||
unit := map[string]any{"unit": ControllerUnit}
|
||||
if _, err := p.Call.Call(verb+"@"+node, unit); err != nil {
|
||||
return false, err.Error()
|
||||
}
|
||||
raw, err := p.Call.Call(CallServiceStatus+"@"+node, unit)
|
||||
if err != nil {
|
||||
return false, "its status cannot be read: " + err.Error()
|
||||
}
|
||||
var st struct {
|
||||
Active string `json:"ActiveState"`
|
||||
}
|
||||
_ = json.Unmarshal(unwrap(raw), &st)
|
||||
if verb == CallServiceStart {
|
||||
return st.Active == "active", "it reads " + st.Active
|
||||
}
|
||||
return st.Active == "inactive" || st.Active == "failed", "it reads " + st.Active
|
||||
}
|
||||
|
||||
// loaded says the bus node's have names the list of that digest loaded from the grants part's key.
|
||||
func (p *Part) loaded(digest string) bool {
|
||||
l := ReadInputs(p.Read, &p.cache).Loaded
|
||||
return digest != "" && l != nil && l.Source == delivery.GrantsUserListSource && l.Digest == digest
|
||||
}
|
||||
|
||||
// setExpiry asks the bus module's tool to set the lease bucket's expiry, and reads it back from the bucket.
|
||||
func (p *Part) setExpiry(maxAge string, a LeaseAct) (bool, string) {
|
||||
raw, err := p.Call.Call(CallBucketExpiry, map[string]any{"bucket": buscompose.LeaseBucket, "max_age": maxAge,
|
||||
"why": a.Act + ": " + a.Why, "warrant": a.Ask})
|
||||
if err != nil {
|
||||
return false, err.Error()
|
||||
}
|
||||
var set struct {
|
||||
Now string `json:"now"`
|
||||
}
|
||||
_ = json.Unmarshal(unwrap(raw), &set)
|
||||
if set.Now != maxAge {
|
||||
return false, "the tool answered " + set.Now
|
||||
}
|
||||
if p.Lease != nil {
|
||||
got, err := p.Lease.Expiry()
|
||||
want, _ := time.ParseDuration(strings.Replace(maxAge, "none", "0s", 1))
|
||||
if err != nil || got != want {
|
||||
return false, fmt.Sprintf("the bucket reads back %s (%v)", got, err)
|
||||
}
|
||||
}
|
||||
return true, ""
|
||||
}
|
||||
|
||||
// putRemovalDue writes removal-due with its time to live.
|
||||
func (p *Part) putRemovalDue(ttl time.Duration) error {
|
||||
s, ok := p.Store.(TTLStore)
|
||||
if !ok {
|
||||
return errors.New("the part's store writes no time to live here")
|
||||
}
|
||||
raw, _ := json.Marshal(map[string]any{"due": p.now().Add(ttl), "written": p.now()})
|
||||
_, err := s.PutTTL(StateLease, KeyRemovalDue, raw, ttl)
|
||||
return err
|
||||
}
|
||||
|
||||
// saySeat says one of the seat's events; a failure is logged, never the end of the act.
|
||||
func (p *Part) saySeat(event string, body any) {
|
||||
if p.SaySeat == nil {
|
||||
return
|
||||
}
|
||||
if err := p.SaySeat(event, body); err != nil && p.Logf != nil {
|
||||
p.Logf("the event %s could not be said: %v", event, err)
|
||||
}
|
||||
}
|
||||
|
||||
// leaseStep is the pass's part in the lease write: the open act asked or advanced, the lease watched once handed, and
|
||||
// the removal run when its marker was heard.
|
||||
func (p *Part) leaseStep(res *Result) {
|
||||
for _, step := range []func() error{p.raiseLease, p.advanceLease, p.watchLease} {
|
||||
if err := step(); err != nil {
|
||||
res.Errors = append(res.Errors, err.Error())
|
||||
}
|
||||
}
|
||||
if p.removalHeard.Swap(false) {
|
||||
if err := p.removal(); err != nil {
|
||||
res.Errors = append(res.Errors, err.Error())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// RemovalDueHeard is the follower's: the marker of removal-due was heard (its delete, by expiry or by a hand-back).
|
||||
func (p *Part) RemovalDueHeard() { p.removalHeard.Store(true) }
|
||||
|
||||
// watchLease says lease-altered once when the key or the bucket's expiry differs from what the part wrote (ADR 0582
|
||||
// clause 4); a hand-back under way is the part's own change and is not watched.
|
||||
func (p *Part) watchLease() error {
|
||||
r, err := p.readLease()
|
||||
if err != nil || r.Handed.IsZero() || r.Altered != "" || (r.Open != nil && r.Open.Act == ActLeaseHandBack && r.Open.Step != StepAsked) {
|
||||
return err
|
||||
}
|
||||
var what []string
|
||||
raw, found, err := p.Read.Get(ReadLease, LeaseKey)
|
||||
var h buscompose.LeaseHolder
|
||||
switch {
|
||||
case err != nil:
|
||||
return nil // not read is no change: the next pass reads it again
|
||||
case !found:
|
||||
what = append(what, "the key holder is gone")
|
||||
case json.Unmarshal(raw, &h) != nil || h.HandedTo != Module || h.Ask != r.Ask:
|
||||
what = append(what, "the key holder names "+firstOf(h.HandedTo, h.Instance, "something else")+", not mesh-grants by "+r.Ask)
|
||||
}
|
||||
if p.Lease != nil {
|
||||
if d, err := p.Lease.Expiry(); err == nil && d != 0 {
|
||||
what = append(what, "the lease's bucket expires after "+d.String())
|
||||
}
|
||||
}
|
||||
if len(what) == 0 {
|
||||
return nil
|
||||
}
|
||||
r.Altered = strings.Join(what, "; ")
|
||||
p.saySeat(EventLeaseAltered, map[string]any{"altered": r.Altered, "ask": r.Ask, "at": p.now()})
|
||||
return p.writeLease(r)
|
||||
}
|
||||
|
||||
func firstOf(s ...string) string {
|
||||
for _, x := range s {
|
||||
if x != "" {
|
||||
return x
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// removal is the day-later removal, started by the marker of removal-due alone (ADR 0582 clause 6): when the lease still
|
||||
// names mesh-grants, the controller's consumer of DEAD_LETTER_NOTICES and the stream ASSIGNMENTS are removed, each read
|
||||
// back by the bus module's tool and recorded as a hand act. A failure is said as removal-failed and removal-due is
|
||||
// written again with a day.
|
||||
func (p *Part) removal() error {
|
||||
r, err := p.readLease()
|
||||
if err != nil || r.Handed.IsZero() || !r.Removed.IsZero() {
|
||||
return err // handed back (a hand-back deletes the key), or done already
|
||||
}
|
||||
if raw, found, err := p.Read.Get(ReadLease, LeaseKey); err != nil || !found {
|
||||
return p.removalFailed(r, "the lease cannot be read, so it is not known to name mesh-grants")
|
||||
} else if away, _ := buscompose.LeaseHandedAway(raw, true, nil); !away {
|
||||
return p.removalFailed(r, "the lease no longer names mesh-grants")
|
||||
}
|
||||
why := "a day after the lease write " + r.Ask + " (ADR 0582 clause 6)"
|
||||
var failed []string
|
||||
if raw, err := p.Call.Call(CallConsumerRemove, map[string]any{"stream": buscompose.DeadLetterNoticesStream,
|
||||
"consumer": buscompose.ControllerName, "why": why}); err != nil && !strings.Contains(err.Error(), "has no consumer") {
|
||||
failed = append(failed, "the consumer controller of DEAD_LETTER_NOTICES: "+err.Error())
|
||||
} else if err == nil && !removed(raw) {
|
||||
failed = append(failed, "the consumer controller of DEAD_LETTER_NOTICES does not read back removed")
|
||||
} else {
|
||||
p.recordHandAct("removed the controller's consumer of DEAD_LETTER_NOTICES", why)
|
||||
}
|
||||
if raw, err := p.Call.Call(CallStreamRemove, map[string]any{"stream": buscompose.AssignmentsStream, "why": why}); err != nil {
|
||||
failed = append(failed, "the stream ASSIGNMENTS: "+err.Error())
|
||||
} else if !removed(raw) {
|
||||
failed = append(failed, "the stream ASSIGNMENTS does not read back removed")
|
||||
} else {
|
||||
p.recordHandAct("removed the stream ASSIGNMENTS", why)
|
||||
}
|
||||
if len(failed) > 0 {
|
||||
return p.removalFailed(r, strings.Join(failed, "; "))
|
||||
}
|
||||
r.Removed = p.now()
|
||||
return p.writeLease(r)
|
||||
}
|
||||
|
||||
// removed reads a removal tool's answer: removed now, or not there to remove.
|
||||
func removed(raw json.RawMessage) bool {
|
||||
var a struct {
|
||||
Removed bool `json:"removed"`
|
||||
Present *bool `json:"present"`
|
||||
}
|
||||
_ = json.Unmarshal(unwrap(raw), &a)
|
||||
return a.Removed || (a.Present != nil && !*a.Present)
|
||||
}
|
||||
|
||||
func (p *Part) removalFailed(r LeaseRecord, why string) error {
|
||||
p.saySeat(EventRemovalFailed, map[string]any{"failed": why, "ask": r.Ask, "at": p.now()})
|
||||
if err := p.putRemovalDue(RemovalAfter); err != nil {
|
||||
return fmt.Errorf("the removal failed (%s) and removal-due could not be written again: %v", why, err)
|
||||
}
|
||||
return fmt.Errorf("the day-later removal failed, tried again in a day: %s", why)
|
||||
}
|
||||
|
||||
// recordHandAct records one removal in the hand-act log; said, never fatal.
|
||||
func (p *Part) recordHandAct(what, why string) {
|
||||
if p.Call == nil {
|
||||
return
|
||||
}
|
||||
if _, err := p.Call.Call(CallRecordHandAct, map[string]any{"what": what, "why": why, "cause": "lease-removal"}); err != nil && p.Logf != nil {
|
||||
p.Logf("%s, and the hand act could not be recorded: %v", what, err)
|
||||
}
|
||||
}
|
||||
|
||||
// RemovalDue is the tool grants_removal_due: removal-due written again with a short time to live, so its marker starts
|
||||
// the removal (the recovery for a marker missed while the part was down). Recorded as a hand act.
|
||||
func (p *Part) RemovalDue(caller, why, ttl string) (any, error) {
|
||||
if strings.TrimSpace(why) == "" {
|
||||
return nil, errors.New("grants_removal_due needs a why, kept with the hand act")
|
||||
}
|
||||
d, err := time.ParseDuration(ttl)
|
||||
if err != nil || d < time.Second || d > RemovalAfter {
|
||||
return nil, fmt.Errorf("ttl %q is no time to live between 1s and %s", ttl, RemovalAfter)
|
||||
}
|
||||
r, err := p.readLease()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if r.Handed.IsZero() || !r.Removed.IsZero() {
|
||||
return nil, errors.New("no removal is due: the lease is not handed to mesh-grants, or the removal ran")
|
||||
}
|
||||
if err := p.putRemovalDue(d); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
p.recordHandAct("wrote removal-due again with "+d.String()+" to live", why+" (by "+caller+")")
|
||||
return map[string]any{"written": KeyRemovalDue, "ttl": d.String(), "says": "its marker starts the removal when it expires"}, nil
|
||||
}
|
||||
|
||||
// LeaseState answers where the lease write stands, for grants_lease_check.
|
||||
func (p *Part) LeaseState() (LeaseRecord, error) { return p.readLease() }
|
||||
@@ -34,7 +34,7 @@ const (
|
||||
)
|
||||
|
||||
// States is every state the part keeps.
|
||||
var States = []string{StateAccounts, StateAssertions, StateGrants, StateMemberships, StateTakeover, StateUserList, StateWithheld}
|
||||
var States = []string{StateAccounts, StateAssertions, StateGrants, StateLease, StateMemberships, StateTakeover, StateUserList, StateWithheld}
|
||||
|
||||
// The events the part says.
|
||||
const (
|
||||
@@ -170,9 +170,17 @@ type Part struct {
|
||||
// Forget takes away the record the asker kept of an ask it failed to publish; Nonce is a test's id suffix.
|
||||
Forget func(id string) error
|
||||
Nonce func() string
|
||||
// deciding is set while a take-over is decided, so the composition it makes asks and reconciles nothing.
|
||||
// Lease is the process's hand on the controller's lease (leasewrite.go); nil in the tools, which ask the process.
|
||||
Lease LeaseBus
|
||||
// SaySeat says the seat mesh-grants' events (lease-handed, lease-altered, removal-failed): the process's; nil in the
|
||||
// tools.
|
||||
SaySeat SeatSayer
|
||||
// deciding is set while a take-over is decided, or a lease act composes, so the composition it makes asks and
|
||||
// reconciles nothing.
|
||||
deciding atomic.Bool
|
||||
Now func() time.Time
|
||||
// removalHeard is set when the marker of removal-due was heard, and taken by the next pass (leasewrite.go).
|
||||
removalHeard atomic.Bool
|
||||
Now func() time.Time
|
||||
|
||||
cache Manifests
|
||||
mu sync.Mutex
|
||||
@@ -249,6 +257,8 @@ func (p *Part) Compose() Result {
|
||||
// the trunk names, so the list grants a build's subjects before the want moves to it (issue 774).
|
||||
recs, _ := Records(in)
|
||||
listRecs, left := ListRecords(in)
|
||||
// From the lease write's cut, the controller's write on the lease is taken from the list (ADR 0582 clause 3).
|
||||
listRecs.LeaseCut = p.leaseCut()
|
||||
holdings := HoldingsOf(in)
|
||||
res.RootNotFree = WithRootFree(&listRecs, p.Call)
|
||||
recs.RootFree = listRecs.RootFree
|
||||
@@ -327,6 +337,10 @@ func (p *Part) Compose() Result {
|
||||
members = append(members, m.Membership)
|
||||
}
|
||||
seated := holdings.ByMemberships(members...)
|
||||
// From the lease write the memberships are written to the state alone and no longer said on their subject, which
|
||||
// nothing reads any more (novox/hq ADR 0571 clause 10, issue 774).
|
||||
notice, _ := leaseHandedAway(p.Read)
|
||||
notice = !notice
|
||||
for _, key := range sortedKeys(ms) {
|
||||
// The last membership is the state's own (ADR 0571 clause 7), never the assignments stream's, which the bus
|
||||
// module's tools remove at the lease write (research 072 row 33): a last read there would freeze every membership
|
||||
@@ -346,8 +360,10 @@ func (p *Part) Compose() Result {
|
||||
m.Published, now = "not published (tried again at the next event): its own "+StateMemberships+" cannot be written", false
|
||||
} else {
|
||||
res.MembershipsWritten++
|
||||
if err := p.Publish.PublishMembership(m.Node, m.Module, body); err != nil {
|
||||
res.Errors = append(res.Errors, fmt.Sprintf("the membership of %s is written, its notice on its subject not said: %v", key, err))
|
||||
if notice {
|
||||
if err := p.Publish.PublishMembership(m.Node, m.Module, body); err != nil {
|
||||
res.Errors = append(res.Errors, fmt.Sprintf("the membership of %s is written, its notice on its subject not said: %v", key, err))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -410,10 +426,18 @@ func (p *Part) Compose() Result {
|
||||
if err := p.Reconcile(); err != nil {
|
||||
res.Errors = append(res.Errors, err.Error())
|
||||
}
|
||||
p.leaseStep(&res)
|
||||
}
|
||||
return res
|
||||
}
|
||||
|
||||
// composeQuiet composes as a step of an act: asking and reconciling nothing.
|
||||
func (p *Part) composeQuiet() Result {
|
||||
p.deciding.Store(true)
|
||||
defer p.deciding.Store(false)
|
||||
return p.Compose()
|
||||
}
|
||||
|
||||
// Last is the result of the last pass this part made, kept under the lock alone.
|
||||
func (p *Part) Last() Result {
|
||||
p.mu.Lock()
|
||||
|
||||
@@ -49,7 +49,8 @@ func HeardFrom(event string) string {
|
||||
const EventAccountIssued = "mesh-vault." + buscompose.AccountIssuedEvent
|
||||
|
||||
// Consumes is every event the part hears, as its manifest's consumes name them.
|
||||
var Consumes = []string{EventConditionRaised, EventConditionCleared, EventKeeperRaised, EventKeeperCleared, EventAccountIssued}
|
||||
var Consumes = []string{EventConditionRaised, EventConditionCleared, EventKeeperRaised, EventKeeperCleared, EventAccountIssued,
|
||||
EventBucketExpirySet}
|
||||
|
||||
// IsAccountIssued says an event heard is the vault's account-issued, by its name or its bus subject.
|
||||
func IsAccountIssued(event string) bool {
|
||||
|
||||
@@ -95,6 +95,8 @@ func (p *Part) writeSealed(res *Result, list []byte, users int, leftOut, dropped
|
||||
cur, _, had := p.Sealed()
|
||||
next := cur
|
||||
next.Follow = followOf(p.SwitchState())
|
||||
// The bus node follows this list alone once the lease is handed (ADR 0571 clause 8, ADR 0582 clause 3).
|
||||
next.LeaseHanded = p.leaseHanded()
|
||||
key := ""
|
||||
if in.Loaded != nil {
|
||||
key = in.Loaded.Key
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
@@ -131,8 +132,15 @@ func (p *Part) SwitchState() string {
|
||||
// the refusals of Decide. Asking again while an ask is open is refused.
|
||||
func (p *Part) TakeOver(caller, why string) (any, error) { return p.request(ActTakeOver, caller, why) }
|
||||
|
||||
// HandBack is the tool: it asks the operator to switch the bus node back to the controller's list.
|
||||
func (p *Part) HandBack(caller, why string) (any, error) { return p.request(ActHandBack, caller, why) }
|
||||
// HandBack is the tool: it asks the operator to switch the bus node back to the controller's list. Once the lease is
|
||||
// handed to the grants part it is answered as gone: the grants part's list is the only one (ADR 0547 §11).
|
||||
func (p *Part) HandBack(caller, why string) (any, error) {
|
||||
if p.leaseHanded() {
|
||||
return nil, errors.New("gone: the controller's lease is handed to mesh-grants, so the grants part's list is the only " +
|
||||
"one and the user list is not handed back (ADR 0547 §11); lease-hand-back gives the lease back first")
|
||||
}
|
||||
return p.request(ActHandBack, caller, why)
|
||||
}
|
||||
|
||||
// nonce is four hex digits that keep two asks of one second apart.
|
||||
func (p *Part) nonce() string {
|
||||
@@ -340,6 +348,9 @@ func (p *Part) AskOf(r Request, st UserListState) (asks.Ask, error) {
|
||||
// Warranted is the process's act on the operator's answer (asker.Client.Listen hands it once, after checking the ask, the
|
||||
// option and its level). A no is recorded; a yes runs the switch after the refusals, and a refusal is said to the operator.
|
||||
func (p *Part) Warranted(o asks.Option, w asks.Warrant) error {
|
||||
if isLeaseAsk(w.Ask) {
|
||||
return p.leaseWarranted(o, w)
|
||||
}
|
||||
s, err := p.readSwitch()
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -397,6 +408,13 @@ func (p *Part) Said(h asker.Handled, w asks.Warrant, err error) {
|
||||
if err != nil && p.Logf != nil {
|
||||
p.Logf("the answer to %s (%s) ended in an error: %v", w.Ask, h, err)
|
||||
}
|
||||
if isLeaseAsk(w.Ask) {
|
||||
if r, err := p.readLease(); err == nil && r.Open != nil && r.Open.Ask == w.Ask && r.Open.Step == StepAsked {
|
||||
r.end(string(w.Outcome), string(w.Outcome)+" ("+string(h)+"): nothing was written", p.now())
|
||||
_ = p.writeLease(r)
|
||||
}
|
||||
return
|
||||
}
|
||||
s, err := p.readSwitch()
|
||||
if err != nil || s.Open != w.Ask {
|
||||
return
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"slices"
|
||||
"strings"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/membership"
|
||||
)
|
||||
@@ -29,7 +30,10 @@ const (
|
||||
|
||||
// Tools is every tool, in the manifest's order.
|
||||
var Tools = []string{ToolGrants, ToolMembership, ToolUserList, ToolComposeNow, ToolAssert, ToolTakeOver, ToolHandBack, ToolTakeOverState,
|
||||
ToolAccounts, ToolConsumerMatch, ToolAcceptHashes, ToolLeaseCheck}
|
||||
ToolAccounts, ToolConsumerMatch, ToolAcceptHashes, ToolLeaseCheck, ToolRemovalDue}
|
||||
|
||||
// SeatTool is the tool serving one of the seat mesh-grants' verbs, `mesh-grants.<verb>`.
|
||||
func SeatTool(verb string) string { return catalogue.GrantsSeat + "." + verb }
|
||||
|
||||
// WritesGrants is what grants says of the part's own grants: written once the list the bus loaded carries a build's
|
||||
// accounts, and the one key the assigner reads (ADR 0571 clause 9, which changes ADR 0547 clause 5).
|
||||
|
||||
@@ -0,0 +1,459 @@
|
||||
package manifest
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
"git.novox.be/novox/mesh-sdk/go/seal"
|
||||
|
||||
"mesh-grants/internal/grants"
|
||||
)
|
||||
|
||||
// leaseFake is the bus as the lease write meets it: fakeBus's states and tools, and the seats and tools the write asks.
|
||||
// Each answer is shaped as its writer's code answers it (issue 774): node-service-manager's status the systemd module's
|
||||
// Manager.Status (ActiveState), nats_bucket_expiry nats-tools' BucketExpirySet (now), nats_consumer_remove its
|
||||
// ConsumerRemoved (removed) and an absent consumer its error ("has no consumer"), nats_stream_remove its StreamRemoved
|
||||
// (removed, present), issue-tracker.get mesh-issues' Brief (status), node_trusts mesh-delivery-view's TrustsAnswer
|
||||
// (every), mesh-config's compare its Comparison (source, count), mesh-wants' upgrade compare the assigner's answer
|
||||
// (empty), vault_status mesh-vault's answer (ready, minting.on, minting.held_back).
|
||||
type leaseFake struct {
|
||||
*fakeBus
|
||||
answer map[string]func(map[string]any) (any, error)
|
||||
active string // the controller's ActiveState
|
||||
expiry time.Duration // the lease bucket's
|
||||
ttls map[string]time.Duration
|
||||
log []string // the write's acts on the bus, in order
|
||||
}
|
||||
|
||||
func newLeaseFake(t *testing.T) (*leaseFake, *grants.Part) {
|
||||
t.Helper()
|
||||
b := fixture()
|
||||
b.manifestOf("nats", `{"module":"nats","version":"1","claims":[{"name":"mesh-broker"}],"own-secrets":{"broker":"/var/lib/nats/broker"},`+
|
||||
`"tools":["nats_bucket_expiry","nats_consumer_remove","nats_stream_remove"]}`)
|
||||
b.put(grants.ReadLease, grants.LeaseKey, map[string]any{"instance": "controller@a pid 1", "host": "a", "epoch": 60,
|
||||
"taken": "2026-10-30T14:52:10Z"})
|
||||
b.put(grants.ReadTakeover, health.TakeoverKey, health.Takeover{V: 1, Follow: health.FollowKeeper, By: "operator", Why: "the keeper took over"})
|
||||
busKey, err := seal.GenerateKeyPair()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b.put(grants.ReadHave, "a.nats", delivery.Have{V: delivery.Version, Phase: delivery.PhaseRunning, Health: delivery.HealthOK,
|
||||
UserList: &delivery.UserListWritten{Source: "a.nats.src2", Revision: 7, Digest: "sha256:aa", Users: 6, Key: busKey.PublicKey}})
|
||||
// The controller's hash in the vault, so the list carries the controller and the cut changes it.
|
||||
b.put(grants.ReadAccounts, buscompose.ControllerName, map[string]any{"hash": "$2a$11$controller", "generation": 1,
|
||||
"at": "2026-10-30T14:52:10Z"})
|
||||
f := &leaseFake{fakeBus: b, active: "active", expiry: 15 * time.Second, ttls: map[string]time.Duration{}}
|
||||
f.answer = map[string]func(map[string]any) (any, error){
|
||||
grants.CallIssue: func(map[string]any) (any, error) { return map[string]any{"status": "resolved"}, nil },
|
||||
grants.CallNodeTrusts: func(map[string]any) (any, error) { return map[string]any{"every": true, "not_yet": []string{}}, nil },
|
||||
grants.CallConfigCompare: func(map[string]any) (any, error) { return map[string]any{"source": "indexer", "count": 0}, nil },
|
||||
grants.CallUpgrade: func(map[string]any) (any, error) { return map[string]any{"empty": true, "splits": []any{}}, nil },
|
||||
grants.CallVaultStatus: func(map[string]any) (any, error) {
|
||||
return map[string]any{"ready": true, "minting": map[string]any{"on": true, "held_back": map[string]string{}}}, nil
|
||||
},
|
||||
grants.CallServiceStop: func(map[string]any) (any, error) { f.active = "inactive"; return map[string]any{"ok": true}, nil },
|
||||
grants.CallServiceStart: func(map[string]any) (any, error) { f.active = "active"; return map[string]any{"ok": true}, nil },
|
||||
grants.CallServiceStatus: func(map[string]any) (any, error) { return map[string]any{"ActiveState": f.active}, nil },
|
||||
grants.CallBucketExpiry: func(a map[string]any) (any, error) {
|
||||
now := fmt.Sprint(a["max_age"])
|
||||
d, _ := time.ParseDuration(strings.Replace(now, "none", "0s", 1))
|
||||
f.expiry = d
|
||||
f.log = append(f.log, "expiry "+now)
|
||||
return map[string]any{"bucket": a["bucket"], "now": now, "changed": true}, nil
|
||||
},
|
||||
grants.CallRecordWarrant: func(a map[string]any) (any, error) { f.log = append(f.log, "warranted"); return map[string]any{}, nil },
|
||||
grants.CallRecordHandAct: func(a map[string]any) (any, error) {
|
||||
f.log = append(f.log, "hand-act "+fmt.Sprint(a["what"]))
|
||||
return map[string]any{}, nil
|
||||
},
|
||||
grants.CallConsumerRemove: func(map[string]any) (any, error) {
|
||||
f.log = append(f.log, "consumer removed")
|
||||
return map[string]any{"removed": true}, nil
|
||||
},
|
||||
grants.CallStreamRemove: func(map[string]any) (any, error) {
|
||||
f.log = append(f.log, "stream removed")
|
||||
return map[string]any{"removed": true, "present": true}, nil
|
||||
},
|
||||
}
|
||||
p := partOn(b)
|
||||
p.Call, p.Store, p.Lease, p.Asker, p.Objects = f, ttlStore{ownStore{b}, f}, f, &fakeAsker{wheres: map[string]*asks.Warrant{}}, &fakeObjects{}
|
||||
p.SaySeat = func(event string, _ any) error { f.log = append(f.log, "event "+event); return nil }
|
||||
p.Nonce = func() string { return "0001" }
|
||||
return f, p
|
||||
}
|
||||
|
||||
func (f *leaseFake) Call(key string, body any) (json.RawMessage, error) {
|
||||
base, _, _ := strings.Cut(key, "@")
|
||||
if answer, ok := f.answer[base]; ok {
|
||||
f.mu.Lock()
|
||||
f.calls = append(f.calls, key)
|
||||
f.mu.Unlock()
|
||||
args, _ := body.(map[string]any)
|
||||
out, err := answer(args)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return json.Marshal(out)
|
||||
}
|
||||
return f.fakeBus.Call(key, body)
|
||||
}
|
||||
|
||||
func (f *leaseFake) PutHolder(value []byte) error {
|
||||
f.log = append(f.log, "holder")
|
||||
f.states[grants.ReadLease][grants.LeaseKey] = value
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *leaseFake) DeleteHolder() error {
|
||||
f.log = append(f.log, "holder deleted")
|
||||
delete(f.states[grants.ReadLease], grants.LeaseKey)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *leaseFake) Expiry() (time.Duration, error) { return f.expiry, nil }
|
||||
|
||||
// loads says the bus node loaded the list the part last sealed.
|
||||
func (f *leaseFake) loads(p *grants.Part) {
|
||||
g, _, _ := p.Sealed()
|
||||
f.put(grants.ReadHave, "a.nats", delivery.Have{V: delivery.Version, Phase: delivery.PhaseRunning, Health: delivery.HealthOK,
|
||||
UserList: &delivery.UserListWritten{Source: delivery.GrantsUserListSource, Revision: g.Revision, Digest: g.Digest,
|
||||
Users: g.Users, Key: keyOf(f)}})
|
||||
}
|
||||
|
||||
func keyOf(f *leaseFake) string {
|
||||
var h delivery.Have
|
||||
_ = json.Unmarshal(f.states[grants.ReadHave]["a.nats"], &h)
|
||||
return h.UserList.Key
|
||||
}
|
||||
|
||||
// ttlStore is the part's own store with a time to live per key, recorded.
|
||||
type ttlStore struct {
|
||||
ownStore
|
||||
f *leaseFake
|
||||
}
|
||||
|
||||
func (s ttlStore) PutTTL(state, key string, value []byte, ttl time.Duration) (uint64, error) {
|
||||
s.f.ttls[state+"."+key] = ttl
|
||||
s.f.log = append(s.f.log, key+" "+ttl.String())
|
||||
return s.Put(state, key, value)
|
||||
}
|
||||
|
||||
// record is the part's lease record now.
|
||||
func record(t *testing.T, p *grants.Part) grants.LeaseRecord {
|
||||
t.Helper()
|
||||
r, err := p.LeaseState()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return r
|
||||
}
|
||||
|
||||
// warrant answers the open act's ask with option, as the router's warrant hands it.
|
||||
func warrant(t *testing.T, p *grants.Part, option string) error {
|
||||
t.Helper()
|
||||
r := record(t, p)
|
||||
if r.Open == nil {
|
||||
t.Fatal("no act is open")
|
||||
}
|
||||
return p.Warranted(asks.Option{ID: option}, asks.Warrant{Ask: r.Open.Ask, Option: option, Outcome: asks.OutcomeChosen})
|
||||
}
|
||||
|
||||
// Issue 774 (novox/hq ADR 0582): the lease write, from its ask through the day-later removal, against stand-in seats.
|
||||
func TestTheLeaseWrite(t *testing.T) {
|
||||
// 774/S14/AC1: any one precondition not green refuses the verb naming it; all green asks.
|
||||
t.Run("preconditions", func(t *testing.T) {
|
||||
for _, row := range []struct {
|
||||
name string
|
||||
break_ func(f *leaseFake)
|
||||
}{
|
||||
{"lease-check", func(f *leaseFake) {
|
||||
f.manifestOf("mod-u", `{"module":"mod-u","version":"1","reads":["mesh-controller.conditions"]}`)
|
||||
}},
|
||||
{"earlier-steps", func(f *leaseFake) {
|
||||
f.answer[grants.CallIssue] = func(a map[string]any) (any, error) {
|
||||
if fmt.Sprint(a["number"]) == "1954" {
|
||||
return map[string]any{"status": "fixing"}, nil
|
||||
}
|
||||
return map[string]any{"status": "resolved"}, nil
|
||||
}
|
||||
}},
|
||||
{"node-trusts", func(f *leaseFake) {
|
||||
f.answer[grants.CallNodeTrusts] = func(map[string]any) (any, error) {
|
||||
return map[string]any{"every": false, "not_yet": []string{"ace"}}, nil
|
||||
}
|
||||
}},
|
||||
{"config-compare", func(f *leaseFake) {
|
||||
f.answer[grants.CallConfigCompare] = func(map[string]any) (any, error) { return map[string]any{"source": "indexer", "count": 2}, nil }
|
||||
}},
|
||||
{"upgrade-compare", func(f *leaseFake) {
|
||||
f.answer[grants.CallUpgrade] = func(map[string]any) (any, error) { return map[string]any{"empty": false, "splits": []any{1}}, nil }
|
||||
}},
|
||||
{"keeper-take-over", func(f *leaseFake) {
|
||||
f.put(grants.ReadTakeover, health.TakeoverKey, health.Takeover{V: 1, Follow: health.FollowController})
|
||||
}},
|
||||
{"vault-minting", func(f *leaseFake) {
|
||||
f.answer[grants.CallVaultStatus] = func(map[string]any) (any, error) {
|
||||
return map[string]any{"ready": true, "minting": map[string]any{"on": true, "held_back": map[string]string{"a.x": "why"}}}, nil
|
||||
}
|
||||
}},
|
||||
{"bus-tools", func(f *leaseFake) { f.manifestOf("nats", natsManifest) }},
|
||||
{"", func(*leaseFake) {}},
|
||||
} {
|
||||
f, p := newLeaseFake(t)
|
||||
row.break_(f)
|
||||
_, err := p.LeaseWrite("operator@shanks", "the controller goes")
|
||||
switch {
|
||||
case row.name == "" && err != nil:
|
||||
t.Errorf("all green: refused: %v", err)
|
||||
case row.name != "" && (err == nil || !strings.Contains(err.Error(), row.name+": ")):
|
||||
t.Errorf("%s not green: %v, want a refusal naming it", row.name, err)
|
||||
case row.name != "":
|
||||
for _, other := range grants.LeaseCheckNames {
|
||||
if other != row.name && strings.Contains(err.Error(), other+": ") {
|
||||
t.Errorf("%s not green: the refusal names %s too: %v", row.name, other, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
if r := record(t, p); (r.Open != nil) != (row.name == "") {
|
||||
t.Errorf("%q: an act open %v", row.name, r.Open)
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
// 774/S15/AC1, AC2: a no, an expiry, or a warrant whose checks changed writes nothing, naming the changed check.
|
||||
t.Run("the ask and the warrant", func(t *testing.T) {
|
||||
for _, row := range []struct {
|
||||
name, step, said string
|
||||
answer func(t *testing.T, f *leaseFake, p *grants.Part)
|
||||
}{
|
||||
{"declined", grants.StepDeclined, "said no", func(t *testing.T, _ *leaseFake, p *grants.Part) {
|
||||
if err := warrant(t, p, "no"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}},
|
||||
{"expired", "expired", "expired", func(t *testing.T, _ *leaseFake, p *grants.Part) {
|
||||
r := record(t, p)
|
||||
p.Said("", asks.Warrant{Ask: r.Open.Ask, Outcome: asks.OutcomeExpired}, nil)
|
||||
}},
|
||||
{"checks changed", grants.StepRefused, "the checks changed since the ask: vault-minting (now: ready true, minting true, 1 held back)",
|
||||
func(t *testing.T, f *leaseFake, p *grants.Part) {
|
||||
// The vault holds one account back between the ask and the warrant.
|
||||
f.answer[grants.CallVaultStatus] = func(map[string]any) (any, error) {
|
||||
return map[string]any{"ready": true, "minting": map[string]any{"on": true, "held_back": map[string]string{"a.x": "why"}}}, nil
|
||||
}
|
||||
if err := warrant(t, p, "yes"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}},
|
||||
} {
|
||||
f, p := newLeaseFake(t)
|
||||
if _, err := p.LeaseWrite("operator@shanks", "the controller goes"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
p.Compose() // the process asks it
|
||||
if a := p.Asker.(*fakeAsker); len(a.asks) != 1 || len(a.asks[0].Options) != 2 || a.asks[0].Options[0].Binds == "" {
|
||||
t.Fatalf("%s: asked %+v, want one ask whose yes binds the act", row.name, a.asks)
|
||||
}
|
||||
row.answer(t, f, p)
|
||||
r := record(t, p)
|
||||
if r.Open != nil || len(r.Acts) != 1 || r.Acts[0].Step != row.step || !strings.Contains(r.Acts[0].Said, row.said) {
|
||||
t.Errorf("%s: open %v, acts %+v; want one act %s saying %q", row.name, r.Open, r.Acts, row.step, row.said)
|
||||
}
|
||||
if f.count(grants.CallServiceStop+"@a") != 0 || len(f.log) != 0 || r.Cut {
|
||||
t.Errorf("%s: something was written: stops %d, %v, cut %v", row.name, f.count(grants.CallServiceStop+"@a"), f.log, r.Cut)
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
// 774/S16/AC1, AC2; 774/S17/AC1, AC2: the write's six steps, in order, each read back before the next.
|
||||
t.Run("the write", func(t *testing.T) {
|
||||
f, p := newLeaseFake(t)
|
||||
f.answer[grants.CallServiceStop] = func(map[string]any) (any, error) { return map[string]any{"ok": true}, nil } // reads back active
|
||||
if _, err := p.LeaseWrite("operator@shanks", "the controller goes"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
p.Compose()
|
||||
if err := warrant(t, p, "yes"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if r := record(t, p); r.Open != nil || r.Cut || r.Acts[0].Step != grants.StepFailed || !strings.Contains(r.Acts[0].Said, "did not read back stopped") || len(f.log) != 0 {
|
||||
t.Fatalf("a stop not read back: %+v, %v; want the act failed there, nothing written", r, f.log)
|
||||
}
|
||||
|
||||
f, p = newLeaseFake(t)
|
||||
if _, err := p.LeaseWrite("operator@shanks", "the controller goes"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
p.Compose()
|
||||
before, _, _ := p.Sealed()
|
||||
if err := warrant(t, p, "yes"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
r := record(t, p)
|
||||
g, _, _ := p.Sealed()
|
||||
if f.active != "inactive" || !r.Cut || r.Open == nil || r.Open.Step != grants.StepCut || r.Open.List != g.Digest || g.Digest == before.Digest {
|
||||
t.Fatalf("after the warrant: active %s, %+v, sealed %s (was %s); want stopped, cut and waiting on the cut list", f.active, r, g.Digest, before.Digest)
|
||||
}
|
||||
if len(f.log) != 0 {
|
||||
t.Fatalf("the expiry or the key was touched before the cut list loaded: %v", f.log)
|
||||
}
|
||||
// The cut list takes the controller's write on the lease and gives mesh-grants the key holder.
|
||||
var st grants.UserListState
|
||||
raw, _, _, _ := ownStore{f.fakeBus}.Get(grants.StateUserList, grants.KeyList)
|
||||
_ = json.Unmarshal(raw, &st)
|
||||
p.Compose() // a pass with the bus node not loaded yet waits
|
||||
if len(f.log) != 0 {
|
||||
t.Fatalf("a pass before the load acted: %v", f.log)
|
||||
}
|
||||
f.loads(p)
|
||||
p.Compose()
|
||||
want := []string{"expiry none", "holder", "event lease-handed", "warranted", "removal-due 24h0m0s"}
|
||||
if !slices.Equal(f.log, want) {
|
||||
t.Errorf("the write's acts %v, want %v in that order", f.log, want)
|
||||
}
|
||||
var h buscompose.LeaseHolder
|
||||
_ = json.Unmarshal(f.states[grants.ReadLease][grants.LeaseKey], &h)
|
||||
r = record(t, p)
|
||||
if h.HandedTo != "mesh-grants" || h.HandedBy != "operator" || h.Why != "the controller goes" || h.Ask != r.Ask || h.At.IsZero() ||
|
||||
r.Open != nil || r.Handed.IsZero() || r.Acts[len(r.Acts)-1].Step != grants.StepDone {
|
||||
t.Errorf("holder %+v, record %+v", h, r)
|
||||
}
|
||||
p.Compose() // the pass the lease key's change wakes
|
||||
if g, _, _ := p.Sealed(); !g.LeaseHanded {
|
||||
t.Error("the user-list key does not say the lease is handed")
|
||||
}
|
||||
if _, err := p.HandBack("operator@shanks", "back"); err == nil || !strings.HasPrefix(err.Error(), "gone") {
|
||||
t.Errorf("the user list's hand-back once the lease is handed: %v, want gone", err)
|
||||
}
|
||||
if _, err := p.LeaseWrite("operator@shanks", "again"); err == nil {
|
||||
t.Error("a second lease-write with the lease handed was asked")
|
||||
}
|
||||
|
||||
// 774/S18/AC1: the key or the expiry changed by another hand is said once, and nothing is written.
|
||||
f.log = nil
|
||||
f.put(grants.ReadLease, grants.LeaseKey, map[string]any{"instance": "controller@a pid 9", "host": "a", "epoch": 61})
|
||||
p.Compose()
|
||||
p.Compose()
|
||||
f.expiry = 15 * time.Second
|
||||
p.Compose()
|
||||
if !slices.Equal(f.log, []string{"event lease-altered"}) || !strings.Contains(record(t, p).Altered, "controller@a pid 9") {
|
||||
t.Errorf("an altered lease said %v (%q), want lease-altered once naming it", f.log, record(t, p).Altered)
|
||||
}
|
||||
})
|
||||
|
||||
// 774/S19/AC1, AC2: hand-back restores the expiry, deletes holder and removal-due, and starts the controller only once
|
||||
// the list with its rights is loaded; once the removal ran it is refused.
|
||||
t.Run("hand-back", func(t *testing.T) {
|
||||
f, p := handed(t)
|
||||
if _, err := p.LeaseHandBack("operator@shanks", "it goes back"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
p.Compose()
|
||||
f.log = nil
|
||||
if err := warrant(t, p, "yes"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
r := record(t, p)
|
||||
if f.expiry != 15*time.Second || f.states[grants.ReadLease][grants.LeaseKey] != nil || r.Cut || f.active != "inactive" {
|
||||
t.Fatalf("after the warrant: expiry %s, holder %s, cut %v, controller %s", f.expiry, f.states[grants.ReadLease][grants.LeaseKey], r.Cut, f.active)
|
||||
}
|
||||
if _, _, found, _ := f.ownGet(grants.StateLease, grants.KeyRemovalDue); found {
|
||||
t.Error("removal-due is not deleted")
|
||||
}
|
||||
f.loads(p)
|
||||
p.Compose()
|
||||
if r := record(t, p); f.active != "active" || r.Open != nil || r.Acts[len(r.Acts)-1].Step != grants.StepDone {
|
||||
t.Errorf("once loaded: controller %s, %+v", f.active, r)
|
||||
}
|
||||
// AC2: once the removal ran, refused naming why.
|
||||
f, p = handed(t)
|
||||
p.RemovalDueHeard()
|
||||
p.Compose()
|
||||
if _, err := p.LeaseHandBack("operator@shanks", "too late"); err == nil || !strings.Contains(err.Error(), "without its consumer") {
|
||||
t.Errorf("hand-back after the removal: %v", err)
|
||||
}
|
||||
})
|
||||
|
||||
// 774/S20/AC1, AC2, AC3: the marker of removal-due starts the removal; a failure says removal-failed and writes the key
|
||||
// again with a day; a hand-back's delete removes nothing.
|
||||
t.Run("the day-later removal", func(t *testing.T) {
|
||||
f, p := handed(t)
|
||||
f.log = nil
|
||||
p.RemovalDueHeard()
|
||||
p.Compose()
|
||||
want := []string{"consumer removed", "hand-act removed the controller's consumer of DEAD_LETTER_NOTICES", "stream removed",
|
||||
"hand-act removed the stream ASSIGNMENTS"}
|
||||
if !slices.Equal(f.log, want) || record(t, p).Removed.IsZero() {
|
||||
t.Errorf("the removal %v (removed %v), want %v", f.log, record(t, p).Removed, want)
|
||||
}
|
||||
|
||||
f, p = handed(t)
|
||||
f.log = nil
|
||||
f.answer[grants.CallStreamRemove] = func(map[string]any) (any, error) {
|
||||
return nil, errors.New("ASSIGNMENTS has 1 consumers: a reader is attached, so it was not removed")
|
||||
}
|
||||
p.RemovalDueHeard()
|
||||
p.Compose()
|
||||
if !slices.Contains(f.log, "event removal-failed") || f.ttls[grants.StateLease+"."+grants.KeyRemovalDue] != grants.RemovalAfter ||
|
||||
!record(t, p).Removed.IsZero() {
|
||||
t.Errorf("a failed removal: %v, removal-due %s", f.log, f.ttls[grants.StateLease+"."+grants.KeyRemovalDue])
|
||||
}
|
||||
|
||||
f, p = handed(t)
|
||||
if _, err := p.LeaseHandBack("operator@shanks", "it goes back"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
p.Compose()
|
||||
if err := warrant(t, p, "yes"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
f.log = nil
|
||||
p.RemovalDueHeard() // the hand-back's own delete of removal-due, heard
|
||||
p.Compose()
|
||||
if slices.ContainsFunc(f.log, func(s string) bool { return strings.Contains(s, "removed") }) {
|
||||
t.Errorf("a hand-back's delete of removal-due removed %v", f.log)
|
||||
}
|
||||
})
|
||||
|
||||
// 774/S21/AC1: once the lease is handed away a membership is written to the state and not published on its subject.
|
||||
t.Run("memberships after the write", func(t *testing.T) {
|
||||
f, p := handed(t)
|
||||
f.publishes = nil
|
||||
f.assign("b", "mod-x")
|
||||
p.Compose()
|
||||
if len(f.publishes) != 0 {
|
||||
t.Errorf("published on mesh.assignment after the lease write: %v", f.publishes)
|
||||
}
|
||||
if _, _, found, _ := f.ownGet(grants.StateMemberships, "b.mod-x"); !found {
|
||||
t.Error("the membership is not written to the state")
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// handed is a part whose lease write is done: the lease names mesh-grants and removal-due is written.
|
||||
func handed(t *testing.T) (*leaseFake, *grants.Part) {
|
||||
t.Helper()
|
||||
f, p := newLeaseFake(t)
|
||||
if _, err := p.LeaseWrite("operator@shanks", "the controller goes"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
p.Compose()
|
||||
if err := warrant(t, p, "yes"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
f.loads(p)
|
||||
p.Compose()
|
||||
if r := record(t, p); r.Handed.IsZero() {
|
||||
t.Fatalf("the write did not finish: %+v, %v", r, f.log)
|
||||
}
|
||||
return f, p
|
||||
}
|
||||
@@ -2,6 +2,16 @@
|
||||
"module": "mesh-grants",
|
||||
"version": "1",
|
||||
"slug": "grants",
|
||||
"claims": [
|
||||
{
|
||||
"name": "mesh-grants",
|
||||
"scope": "mesh",
|
||||
"serves": [
|
||||
"lease-write",
|
||||
"lease-hand-back"
|
||||
]
|
||||
}
|
||||
],
|
||||
"reads": [
|
||||
"mesh-config.resolved",
|
||||
"mesh-controller.assignments",
|
||||
@@ -20,7 +30,8 @@
|
||||
"mesh-controller.condition-cleared",
|
||||
"mesh-conditions.condition-raised",
|
||||
"mesh-conditions.condition-cleared",
|
||||
"mesh-vault.account-issued"
|
||||
"mesh-vault.account-issued",
|
||||
"nats.bucket-expiry-set"
|
||||
],
|
||||
"uses": [
|
||||
"operator-channel"
|
||||
@@ -30,7 +41,19 @@
|
||||
"nats.nats_user_can",
|
||||
"nats.nats_users",
|
||||
"seat:node-engine.root-free",
|
||||
"seat:mesh-hand-acts.warranted"
|
||||
"seat:mesh-hand-acts.warranted",
|
||||
"seat:node-service-manager.stop",
|
||||
"seat:node-service-manager.start",
|
||||
"seat:node-service-manager.status",
|
||||
"nats.nats_bucket_expiry",
|
||||
"nats.nats_consumer_remove",
|
||||
"nats.nats_stream_remove",
|
||||
"seat:mesh-hand-acts.hand-act",
|
||||
"mesh-delivery-view.node_trusts",
|
||||
"seat:mesh-config.compare",
|
||||
"seat:mesh-wants.upgrade",
|
||||
"mesh-vault.vault_status",
|
||||
"seat:issue-tracker.get"
|
||||
],
|
||||
"emits": [
|
||||
"asserted",
|
||||
@@ -51,7 +74,8 @@
|
||||
"grants_accounts",
|
||||
"grants_consumer_match",
|
||||
"grants_accept_hashes",
|
||||
"grants_lease_check"
|
||||
"grants_lease_check",
|
||||
"grants_removal_due"
|
||||
],
|
||||
"state": [
|
||||
{
|
||||
@@ -66,6 +90,10 @@
|
||||
"name": "grants",
|
||||
"history": 5
|
||||
},
|
||||
{
|
||||
"name": "lease",
|
||||
"key-ttl": true
|
||||
},
|
||||
{
|
||||
"name": "memberships",
|
||||
"history": 5
|
||||
|
||||
Reference in New Issue
Block a user