Merge remote-tracking branch 'origin/main' into adr-0297-module-logs
This commit is contained in:
@@ -3,6 +3,7 @@ package inventory
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
@@ -172,6 +173,9 @@ func declaredFor(m catalogue.Manifest, seats map[string]catalogue.SeatDeclaratio
|
||||
Checks: catalogue.HealthChecks(m),
|
||||
// And whether it runs as an account of its own (novox/hq ADR 0259 §8).
|
||||
RunsAs: m.RunsAs,
|
||||
// And the tools that answer a credential its software issued, which the controller asks in its own
|
||||
// name (novox/hq ADR 0305).
|
||||
Issued: issuedTools(m),
|
||||
}
|
||||
// Whether it can be given an account at all: delivered as its own secret named broker, so one
|
||||
// that declares none has nowhere to read it (novox/hq issue 195).
|
||||
@@ -199,6 +203,17 @@ func declaredFor(m catalogue.Manifest, seats map[string]catalogue.SeatDeclaratio
|
||||
return d
|
||||
}
|
||||
|
||||
// issuedTools are the tools through which a module answers the credentials its software issues (novox/hq ADR 0305).
|
||||
func issuedTools(m catalogue.Manifest) []string {
|
||||
var out []string
|
||||
for _, o := range m.Provides {
|
||||
if tool, issued := m.IssuedCredentialOf(o.Name); issued && !slices.Contains(out, tool) {
|
||||
out = append(out, tool)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// bucketsOf is the state a module keeps, as the bus holds it.
|
||||
func bucketsOf(m catalogue.Manifest) []broker.Bucket {
|
||||
var out []broker.Bucket
|
||||
|
||||
@@ -0,0 +1,155 @@
|
||||
package inventory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/secrets"
|
||||
)
|
||||
|
||||
// A credential a provider's software issued, kept as the provision's one credential (novox/hq ADR 0305, issue
|
||||
// 526).
|
||||
//
|
||||
// **The software made it, so the mesh carries it.** A media server's sign-in token comes from its vendor; the mesh
|
||||
// can neither make nor rotate it. The controller fetches it from the provider's own tool, sealed to a key made for
|
||||
// that one call, and hands the value here, which seals it to every consumer that holds the provision from that
|
||||
// provider — in the pair credential's shape, so a consumer's definition does not know where it came from (ADR 0158
|
||||
// §5) — and to the operator, and forgets it.
|
||||
//
|
||||
// **Asked often, written seldom.** The provider asks hourly and on every start. The mesh keeps no plaintext, so it
|
||||
// cannot compare values; it stamps every row with the value's fingerprint instead, and a value whose stamp every
|
||||
// row already carries is not sealed again — sealing anew would change every consumer's declaration and restart
|
||||
// each of them for nothing.
|
||||
|
||||
// OriginIssued is a pair credential the provider's software issued and the provider gave (novox/hq ADR 0305).
|
||||
const OriginIssued = "issued"
|
||||
|
||||
// IssuedStamp is the generation a value issued by a provider's software is kept under: a fingerprint of the value,
|
||||
// so the same value fetched again is known as the same without being kept.
|
||||
func IssuedStamp(value string) string {
|
||||
sum := sha256.Sum256([]byte(value))
|
||||
return "issued:sha256:" + hex.EncodeToString(sum[:8])
|
||||
}
|
||||
|
||||
// IssuedKept is what keeping an issued value did: whether it was sealed anew, the pairs that hold it, and the
|
||||
// machines a send must reach — the provider's and every holder's — when it was.
|
||||
type IssuedKept struct {
|
||||
Changed bool
|
||||
// Holders are every pair that holds the provision from the provider; Sealed those sealed anew by this act.
|
||||
Holders []Holder
|
||||
Sealed []Holder
|
||||
Machines []string
|
||||
// Unsealable are the holders whose machine has no sealing key: left as they are, said by name.
|
||||
Unsealable []Holder
|
||||
}
|
||||
|
||||
// KeepIssuedSecret seals a value a provider's software issued to every pair of provision from provider that does
|
||||
// not hold it yet, under one stamp; a pair that does is left as it is, and when every pair does nothing is written
|
||||
// (novox/hq ADR 0305 §4, §5).
|
||||
func (i *Inventory) KeepIssuedSecret(ctx context.Context, provision, provider, value string) (IssuedKept, error) {
|
||||
if strings.TrimSpace(value) == "" {
|
||||
return IssuedKept{}, fmt.Errorf("there is nothing to seal")
|
||||
}
|
||||
providerNode, err := i.NodeByName(ctx, provider)
|
||||
if err != nil {
|
||||
return IssuedKept{}, err
|
||||
}
|
||||
providerKey, err := i.SealingKeyOf(ctx, provider)
|
||||
if err != nil {
|
||||
return IssuedKept{}, err
|
||||
}
|
||||
if providerKey == "" {
|
||||
return IssuedKept{}, fmt.Errorf("%s has no sealing key, so %s cannot be sealed to it", provider, provision)
|
||||
}
|
||||
stamp := IssuedStamp(value)
|
||||
|
||||
type row struct {
|
||||
holder Holder
|
||||
consumerID any
|
||||
generation, heldC, heldP string
|
||||
consumerKey string
|
||||
}
|
||||
rows, err := i.store.Pool().Query(ctx,
|
||||
`select s.consumer, c.name, s.consumer_module, s.local, coalesce(s.generation, ''),
|
||||
coalesce(s.consumer_key, ''), coalesce(s.provider_key, ''), coalesce(c.sealing_key, '')
|
||||
from secret s join node c on c.id = s.consumer
|
||||
where s.provider = $1 and s.name = $2
|
||||
order by c.name, s.consumer_module, s.local`, providerNode.ID, provision)
|
||||
if err != nil {
|
||||
return IssuedKept{}, err
|
||||
}
|
||||
var held []row
|
||||
for rows.Next() {
|
||||
var r row
|
||||
if err := rows.Scan(&r.consumerID, &r.holder.Consumer, &r.holder.ConsumerModule, &r.holder.Local,
|
||||
&r.generation, &r.heldC, &r.heldP, &r.consumerKey); err != nil {
|
||||
rows.Close()
|
||||
return IssuedKept{}, err
|
||||
}
|
||||
r.holder.Provision, r.holder.Provider = provision, provider
|
||||
held = append(held, r)
|
||||
}
|
||||
rows.Close()
|
||||
if err := rows.Err(); err != nil {
|
||||
return IssuedKept{}, err
|
||||
}
|
||||
|
||||
var kept IssuedKept
|
||||
var stale []row
|
||||
for _, r := range held {
|
||||
if r.consumerKey == "" {
|
||||
kept.Unsealable = append(kept.Unsealable, r.holder)
|
||||
continue
|
||||
}
|
||||
kept.Holders = append(kept.Holders, r.holder)
|
||||
// Only a pair that does not hold this value, sealed to the keys its ends have now, is sealed anew: a
|
||||
// holder that already does is left byte for byte as it is, so it is not restarted for another's sake.
|
||||
if r.generation != stamp || r.heldC != r.consumerKey || r.heldP != providerKey {
|
||||
stale = append(stale, r)
|
||||
}
|
||||
}
|
||||
if len(stale) == 0 {
|
||||
return kept, nil
|
||||
}
|
||||
|
||||
forOperator, operatorKey, err := i.operatorSeal(ctx, value)
|
||||
if err != nil {
|
||||
return IssuedKept{}, err
|
||||
}
|
||||
tx, err := i.store.Pool().Begin(ctx)
|
||||
if err != nil {
|
||||
return IssuedKept{}, err
|
||||
}
|
||||
defer func() { _ = tx.Rollback(ctx) }()
|
||||
machines := map[string]bool{provider: true}
|
||||
for _, r := range stale {
|
||||
sealed, err := secrets.Accept(value, r.consumerKey, providerKey)
|
||||
if err != nil {
|
||||
return IssuedKept{}, err
|
||||
}
|
||||
if _, err := tx.Exec(ctx,
|
||||
`update secret set for_consumer = $6, for_provider = $7, consumer_key = $8, provider_key = $9,
|
||||
operator_sealed = $10, operator_key = $11, origin = $12, generation = $13, created_at = now()
|
||||
where name = $1 and consumer = $2 and consumer_module = $3 and provider = $4 and local = $5`,
|
||||
provision, r.consumerID, r.holder.ConsumerModule, providerNode.ID, r.holder.Local,
|
||||
sealed.ForConsumer, sealed.ForProvider, sealed.ConsumerKey, sealed.ProviderKey,
|
||||
forOperator, operatorKey, OriginIssued, stamp); err != nil {
|
||||
return IssuedKept{}, err
|
||||
}
|
||||
machines[r.holder.Consumer] = true
|
||||
kept.Sealed = append(kept.Sealed, r.holder)
|
||||
}
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return IssuedKept{}, err
|
||||
}
|
||||
kept.Changed = true
|
||||
for m := range machines {
|
||||
kept.Machines = append(kept.Machines, m)
|
||||
}
|
||||
sort.Strings(kept.Machines)
|
||||
return kept, nil
|
||||
}
|
||||
@@ -0,0 +1,178 @@
|
||||
package inventory
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/catalogue"
|
||||
)
|
||||
|
||||
// A value a provider's software issued is sealed to every pair that holds the provision from that provider, under
|
||||
// one stamp of the value, and the same value given again writes nothing and sends nobody (novox/hq ADR 0305 §4,
|
||||
// §5, issue 526).
|
||||
func TestAnIssuedValueIsSealedToEveryHolderOnceAndNotAgain(t *testing.T) {
|
||||
inv, ctx := twoNodesWithKeys(t)
|
||||
consumerKey, consumerOpens := aSealingKey(t)
|
||||
cn, _ := inv.NodeByName(ctx, "consumer")
|
||||
if err := inv.RecordSealingKey(ctx, cn.ID, consumerKey); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, m := range []string{"kometa", "tautulli"} {
|
||||
if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: m, Version: "1", Requires: []string{"plex-api"},
|
||||
Secrets: map[string]string{"plex-api": "/run/plex-api.secret"}}, Source{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// What the mesh made before the provider gave its value: a random value per pair (issue 164).
|
||||
if _, err := inv.SecretFor(ctx, "plex-api", "consumer", m, "provider", ""); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
kept, err := inv.KeepIssuedSecret(ctx, "plex-api", "provider", "the-servers-own-token")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !kept.Changed || len(kept.Holders) != 2 {
|
||||
t.Fatalf("the value was not sealed to both holders: %+v", kept)
|
||||
}
|
||||
if len(kept.Machines) != 2 || kept.Machines[0] != "consumer" || kept.Machines[1] != "provider" {
|
||||
t.Fatalf("the machines to send are the provider's and every holder's: %v", kept.Machines)
|
||||
}
|
||||
for _, m := range []string{"kometa", "tautulli"} {
|
||||
got, err := inv.SecretFor(ctx, "plex-api", "consumer", m, "provider", "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got.Origin != OriginIssued {
|
||||
t.Fatalf("%s's pair says it came from %q, not the provider's software", m, got.Origin)
|
||||
}
|
||||
opened, ok := consumerOpens(got.ForConsumer)
|
||||
if !ok || string(opened) != "the-servers-own-token" {
|
||||
t.Fatalf("%s's copy is not the provider's value", m)
|
||||
}
|
||||
var generation string
|
||||
pn, _ := inv.NodeByName(ctx, "provider")
|
||||
if err := inv.store.Pool().QueryRow(ctx, `select generation from secret where name = 'plex-api' and consumer = $1
|
||||
and consumer_module = $2 and provider = $3`, cn.ID, m, pn.ID).Scan(&generation); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if generation != IssuedStamp("the-servers-own-token") {
|
||||
t.Fatalf("%s's pair is not stamped with the value's fingerprint: %q", m, generation)
|
||||
}
|
||||
}
|
||||
before, _ := inv.SecretFor(ctx, "plex-api", "consumer", "kometa", "provider", "")
|
||||
|
||||
// A consumer that binds later gets a pair of its own, and no holder's value is remade for it: the offer names
|
||||
// no own secret, so ADR 0158 §6's remaking on a bind does not reach it (ADR 0305 §7).
|
||||
if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: "overseerr", Version: "1", Requires: []string{"plex-api"},
|
||||
Secrets: map[string]string{"plex-api": "/run/plex-api.secret"}}, Source{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
late, err := inv.SecretFor(ctx, "plex-api", "consumer", "overseerr", "provider", "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if late.Origin == OriginIssued {
|
||||
t.Fatal("a consumer that bound after the ask holds the issued value without an ask")
|
||||
}
|
||||
if held, _ := inv.SecretFor(ctx, "plex-api", "consumer", "kometa", "provider", ""); held.ForConsumer != before.ForConsumer {
|
||||
t.Fatal("a later bind remade a holder's issued value")
|
||||
}
|
||||
|
||||
// The next ask seals the provider's value to the late consumer alone; the holders that had it are not touched.
|
||||
again, err := inv.KeepIssuedSecret(ctx, "plex-api", "provider", "the-servers-own-token")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !again.Changed || len(again.Sealed) != 1 || again.Sealed[0].ConsumerModule != "overseerr" {
|
||||
t.Fatalf("the late consumer was not sealed to, or not alone: %+v", again)
|
||||
}
|
||||
if held, _ := inv.SecretFor(ctx, "plex-api", "consumer", "kometa", "provider", ""); held.ForConsumer != before.ForConsumer {
|
||||
t.Fatal("sealing the late consumer changed what a holder already held")
|
||||
}
|
||||
if got, _ := inv.SecretFor(ctx, "plex-api", "consumer", "overseerr", "provider", ""); got.Origin != OriginIssued {
|
||||
t.Fatal("the late consumer does not hold the issued value after the ask")
|
||||
}
|
||||
|
||||
// And once every holder holds it, the same value writes nothing and sends nobody.
|
||||
still, err := inv.KeepIssuedSecret(ctx, "plex-api", "provider", "the-servers-own-token")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if still.Changed || len(still.Machines) != 0 {
|
||||
t.Fatalf("the same value was sealed again, which restarts every consumer for nothing: %+v", still)
|
||||
}
|
||||
after, _ := inv.SecretFor(ctx, "plex-api", "consumer", "kometa", "provider", "")
|
||||
if after.ForConsumer != before.ForConsumer {
|
||||
t.Fatal("the same value given again changed what the consumer holds")
|
||||
}
|
||||
|
||||
changed, err := inv.KeepIssuedSecret(ctx, "plex-api", "provider", "a-token-plex-changed")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !changed.Changed {
|
||||
t.Fatal("a changed value was not sealed")
|
||||
}
|
||||
got, _ := inv.SecretFor(ctx, "plex-api", "consumer", "tautulli", "provider", "")
|
||||
if opened, ok := consumerOpens(got.ForConsumer); !ok || string(opened) != "a-token-plex-changed" {
|
||||
t.Fatal("the changed value did not reach the holder")
|
||||
}
|
||||
|
||||
if _, err := inv.KeepIssuedSecret(ctx, "plex-api", "provider", " "); err == nil {
|
||||
t.Fatal("an empty value was kept")
|
||||
}
|
||||
}
|
||||
|
||||
// A consumer whose sealing key changed holds a value the mesh made in place of the issued one, and the next ask
|
||||
// seals the issued value to it again; so does a pair a person accepted over it (novox/hq ADR 0305 §5, the review
|
||||
// of mesh-controller #239).
|
||||
func TestAnIssuedValueIsSealedAgainAfterAKeyChangeOrAnAcceptance(t *testing.T) {
|
||||
inv, ctx := twoNodesWithKeys(t)
|
||||
if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: "kometa", Version: "1", Requires: []string{"plex-api"},
|
||||
Secrets: map[string]string{"plex-api": "/run/plex-api.secret"}}, Source{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := inv.SecretFor(ctx, "plex-api", "consumer", "kometa", "provider", ""); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if kept, err := inv.KeepIssuedSecret(ctx, "plex-api", "provider", "the-servers-own-token"); err != nil || !kept.Changed {
|
||||
t.Fatalf("the first ask did not seal: %+v %v", kept, err)
|
||||
}
|
||||
|
||||
// The consumer rejoins with a new key: the mesh remakes the pair, a value the software refuses.
|
||||
newKey, opens := aSealingKey(t)
|
||||
cn, _ := inv.NodeByName(ctx, "consumer")
|
||||
if err := inv.RecordSealingKey(ctx, cn.ID, newKey); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
remade, err := inv.SecretFor(ctx, "plex-api", "consumer", "kometa", "provider", "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if remade.Origin != OriginMade {
|
||||
t.Fatalf("a value the mesh remade says it came from %q", remade.Origin)
|
||||
}
|
||||
again, err := inv.KeepIssuedSecret(ctx, "plex-api", "provider", "the-servers-own-token")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !again.Changed || len(again.Sealed) != 1 {
|
||||
t.Fatalf("the issued value was not sealed again after the key changed: %+v", again)
|
||||
}
|
||||
got, _ := inv.SecretFor(ctx, "plex-api", "consumer", "kometa", "provider", "")
|
||||
if opened, ok := opens(got.ForConsumer); !ok || string(opened) != "the-servers-own-token" {
|
||||
t.Fatal("the consumer does not hold the issued value under its new key")
|
||||
}
|
||||
|
||||
// A person accepts another value over it: the next ask seals the software's again.
|
||||
if err := inv.AcceptSecretForPair(ctx, "plex-api", "consumer", "kometa", "provider", "", "a-typed-copy"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
after, err := inv.KeepIssuedSecret(ctx, "plex-api", "provider", "the-servers-own-token")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !after.Changed {
|
||||
t.Fatal("an accepted value over the issued one was read as still holding it")
|
||||
}
|
||||
}
|
||||
+142
-48
@@ -111,6 +111,12 @@ type PlanBatch struct {
|
||||
// a merge never shares a batch with one that waits for mesh-delivery's (ADR 0276, decided during the
|
||||
// build, 2026-10-10).
|
||||
Own bool `json:"own,omitempty"`
|
||||
// Closed says the window was closed by a stop that walks a stopped walk's merges again in this batch: they
|
||||
// were heard before, and an older commit never waits behind a newer one (novox/hq ADR 0299). Withheld are
|
||||
// the modules a stopped merge moves: built at the branch, which holds that merge, they would deliver it, so
|
||||
// the batch's walk does not build them.
|
||||
Closed bool `json:"closed,omitempty"`
|
||||
Withheld []string `json:"withheld,omitempty"`
|
||||
}
|
||||
|
||||
// OwnPath says the record is a batch of merges on the controller's own path.
|
||||
@@ -187,6 +193,17 @@ type PlanDelivery struct {
|
||||
// TakenOverBy names the walk a walk folded before it started was taken over by: its merges are that
|
||||
// walk's now.
|
||||
TakenOverBy string `json:"taken_over_by,omitempty"`
|
||||
// WalkedAgainBy names, on a walk stopped without some of its merges, the batch that walks its other merges
|
||||
// again, and LeftOut the merges it left out (novox/hq issue 459, ADR 0299): a delivery of a merge left out
|
||||
// ends stopped with the walk, one of a merge walked again follows that batch's walk.
|
||||
WalkedAgainBy string `json:"walked_again_by,omitempty"`
|
||||
LeftOut []PlanLeftOut `json:"left_out,omitempty"`
|
||||
// Kept are the modules the stopped walk already sent and nothing walks again: the machines named keep running
|
||||
// the stopped walk's build (novox/hq ADR 0299).
|
||||
Kept []PlanKept `json:"kept,omitempty"`
|
||||
// Withheld are, on a walk cut from a batch a stop walked merges again in, the modules it did not build: a
|
||||
// stopped merge moves them, and built only as dependents at the branch they would deliver it (ADR 0299).
|
||||
Withheld []string `json:"withheld,omitempty"`
|
||||
// Batch is the window of a batch; nil once it is cut into a walk.
|
||||
Batch *PlanBatch `json:"batch,omitempty"`
|
||||
// Alone says the walk walks one merge on its own commit, after a failed walk carried it in a later one:
|
||||
@@ -194,6 +211,26 @@ type PlanDelivery struct {
|
||||
Alone bool `json:"alone,omitempty"`
|
||||
}
|
||||
|
||||
// PlanLeftOut is a merge a stop left out of the batch that walks a stopped walk's merges again (novox/hq ADR
|
||||
// 0299): a merge stopped, or a later merge of its repository's branch, which contains the stopped one and so
|
||||
// cannot be walked without delivering it.
|
||||
type PlanLeftOut struct {
|
||||
Repository string `json:"repository"`
|
||||
Commit string `json:"commit"`
|
||||
// Contains is the stopped merge a later merge contains, or whose modules it shares; empty for a merge
|
||||
// stopped itself.
|
||||
Contains string `json:"contains,omitempty"`
|
||||
// Shares are the modules it moves that a stopped merge moves too: built at the branch, which holds the stopped
|
||||
// merge, they would deliver it.
|
||||
Shares []string `json:"shares,omitempty"`
|
||||
}
|
||||
|
||||
// PlanKept is a module a stopped walk sent and nothing walks again, and the machines it reached.
|
||||
type PlanKept struct {
|
||||
Module string `json:"module"`
|
||||
Machines []string `json:"machines"`
|
||||
}
|
||||
|
||||
// Waiting is whether the walk waits for its delivery's word.
|
||||
func (p Plan) Waiting() bool {
|
||||
return p.Delivery != nil && p.Delivery.Awaits != "" && p.Delivery.Go == nil
|
||||
@@ -265,7 +302,10 @@ type PlanGate struct {
|
||||
Sent map[string]SentDeclaration `json:"sent,omitempty"`
|
||||
// Passes counts the consecutive judgings that found it healthy, LastPass the newest; a judging that
|
||||
// does not resets them.
|
||||
Passes int `json:"passes,omitempty"`
|
||||
Passes int `json:"passes,omitempty"`
|
||||
// Wants is how many passes the send needs: a core module's three, a leaf module's one (novox/hq ADR 0282
|
||||
// decision 4 as ADR 0304 decision 3 changed it). Zero in a gate judged before it was said: three.
|
||||
Wants int `json:"wants,omitempty"`
|
||||
LastPass *time.Time `json:"last_pass,omitempty"`
|
||||
// Last is what the newest judging found wanting, while it still may pass.
|
||||
Last string `json:"last,omitempty"`
|
||||
@@ -305,6 +345,18 @@ type PlanGate struct {
|
||||
// Readings are the judging's readings with their times (novox/hq ADR 0282 decision 6): every reading that
|
||||
// counted a pass, and the first that did not after one that did. At most maxReadings, the newest kept.
|
||||
Readings []GateReading `json:"readings,omitempty"`
|
||||
// Grants is the grants step taken before this gate's send (novox/hq issue 490): the bus's user list sent
|
||||
// alone to the machine holding the bus, every build there kept, so what the send newly grants is on the
|
||||
// bus before the gate judges it. Nil when the list did not change, or the gate's machine holds the bus.
|
||||
Grants *GrantsStep `json:"grants,omitempty"`
|
||||
}
|
||||
|
||||
// GrantsStep is the bus's user list sent to the machine holding the bus ahead of a gated send (novox/hq
|
||||
// issue 490): to which machine, when, and, when it could not be sent, why — the send went on without it.
|
||||
type GrantsStep struct {
|
||||
Node string `json:"node"`
|
||||
At *time.Time `json:"at,omitempty"`
|
||||
Failed string `json:"failed,omitempty"`
|
||||
}
|
||||
|
||||
// GateReading is one reading of a first-node gate.
|
||||
@@ -385,39 +437,28 @@ func (p Plan) Open() bool { return p.State == PlanBuilding || p.State == PlanRol
|
||||
// with ErrPlanMoved otherwise; and only by a process that may act (ActsUnder), whose epoch it records.
|
||||
// On success p's revision and epoch are the ones written, so the caller may save it again.
|
||||
func (i *Inventory) SavePlan(ctx context.Context, p *Plan) error {
|
||||
return i.SavePlansMoving(ctx, []*Plan{p}, nil)
|
||||
}
|
||||
|
||||
// MergeMove is a merge given to the plan that answers it now (novox/hq ADR 0276).
|
||||
type MergeMove struct {
|
||||
Repository, Commit, Plan string
|
||||
Alone bool
|
||||
}
|
||||
|
||||
// SavePlansMoving writes plans, as SavePlan does each, and gives merges to the plans that answer them, all in one
|
||||
// transaction: every write and move, or none (novox/hq ADR 0299 decision 7 — a stop that walks merges again never
|
||||
// leaves a merge answered by two plans, nor a walk stopped whose merges were not moved). A plan written by
|
||||
// somebody else since it was read refuses the whole act with ErrPlanMoved. Each plan is said (PlanSaved) once
|
||||
// the transaction is committed, in order.
|
||||
func (i *Inventory) SavePlansMoving(ctx context.Context, plans []*Plan, moves []MergeMove) error {
|
||||
epoch, err := i.actingEpoch(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("the plan for %s %s is not written: %w", p.Repository, p.Commit, err)
|
||||
}
|
||||
tiers, err := json.Marshal(p.Tiers)
|
||||
if err != nil {
|
||||
if len(plans) > 0 {
|
||||
return fmt.Errorf("the plan for %s %s is not written: %w", plans[0].Repository, plans[0].Commit, err)
|
||||
}
|
||||
return err
|
||||
}
|
||||
modules, err := json.Marshal(p.Modules)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var release, delivery, commits, times []byte
|
||||
if p.Times != nil {
|
||||
if times, err = json.Marshal(p.Times); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if len(p.Commits) > 0 {
|
||||
if commits, err = json.Marshal(p.Commits); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if p.Release != nil {
|
||||
if release, err = json.Marshal(p.Release); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if p.Delivery != nil {
|
||||
if delivery, err = json.Marshal(p.Delivery); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
// **And how long the tier it left took** (novox/hq to-be 45 Phase 0): measured here, where the
|
||||
// plan moves, in the same transaction as the move, so no save can move a tier unmeasured or
|
||||
// measure one twice.
|
||||
@@ -426,9 +467,77 @@ func (i *Inventory) SavePlan(ctx context.Context, p *Plan) error {
|
||||
return err
|
||||
}
|
||||
defer func() { _ = tx.Rollback(ctx) }()
|
||||
type written struct {
|
||||
revision int64
|
||||
entered time.Time
|
||||
}
|
||||
done := make([]written, len(plans))
|
||||
for n, p := range plans {
|
||||
revision, entered, err := writePlan(ctx, tx, p, epoch)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
done[n] = written{revision, entered}
|
||||
}
|
||||
for _, m := range moves {
|
||||
if _, err := tx.Exec(ctx,
|
||||
`update batched_merge set plan_id = nullif($3, ''), alone = $4 where repository = lower($1) and commit_hash = $2`,
|
||||
m.Repository, m.Commit, m.Plan, m.Alone); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
for n, p := range plans {
|
||||
p.Revision, p.TierEntered = done[n].revision, done[n].entered
|
||||
if epoch != nil {
|
||||
p.Epoch = uint64(*epoch)
|
||||
} else {
|
||||
p.Epoch = 0
|
||||
}
|
||||
if PlanSaved != nil {
|
||||
PlanSaved(*p)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// writePlan writes one plan within a transaction, by compare-and-set on its revision: its new revision, and when
|
||||
// it entered its tier.
|
||||
func writePlan(ctx context.Context, tx pgx.Tx, p *Plan, epoch *int64) (int64, time.Time, error) {
|
||||
tiers, err := json.Marshal(p.Tiers)
|
||||
if err != nil {
|
||||
return 0, time.Time{}, err
|
||||
}
|
||||
modules, err := json.Marshal(p.Modules)
|
||||
if err != nil {
|
||||
return 0, time.Time{}, err
|
||||
}
|
||||
var release, delivery, commits, times []byte
|
||||
if p.Times != nil {
|
||||
if times, err = json.Marshal(p.Times); err != nil {
|
||||
return 0, time.Time{}, err
|
||||
}
|
||||
}
|
||||
if len(p.Commits) > 0 {
|
||||
if commits, err = json.Marshal(p.Commits); err != nil {
|
||||
return 0, time.Time{}, err
|
||||
}
|
||||
}
|
||||
if p.Release != nil {
|
||||
if release, err = json.Marshal(p.Release); err != nil {
|
||||
return 0, time.Time{}, err
|
||||
}
|
||||
}
|
||||
if p.Delivery != nil {
|
||||
if delivery, err = json.Marshal(p.Delivery); err != nil {
|
||||
return 0, time.Time{}, err
|
||||
}
|
||||
}
|
||||
entered, err := planTierLeft(ctx, tx, *p, time.Now())
|
||||
if err != nil {
|
||||
return err
|
||||
return 0, time.Time{}, err
|
||||
}
|
||||
var revision int64
|
||||
err = tx.QueryRow(ctx,
|
||||
@@ -449,24 +558,9 @@ func (i *Inventory) SavePlan(ctx context.Context, p *Plan) error {
|
||||
// The row is there and at another revision — moved since this was read, or there already
|
||||
// when this one is new: either way not this writer's to overwrite. (A plan saved before plans
|
||||
// had revisions is at zero, and its first save here is from a read at zero.)
|
||||
return fmt.Errorf("the plan for %s %s (%s) is not written: %w", p.Repository, short(p.Commit), p.ID, ErrPlanMoved)
|
||||
return 0, time.Time{}, fmt.Errorf("the plan for %s %s (%s) is not written: %w", p.Repository, short(p.Commit), p.ID, ErrPlanMoved)
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
p.Revision, p.TierEntered = revision, entered
|
||||
if epoch != nil {
|
||||
p.Epoch = uint64(*epoch)
|
||||
} else {
|
||||
p.Epoch = 0
|
||||
}
|
||||
if PlanSaved != nil {
|
||||
PlanSaved(*p)
|
||||
}
|
||||
return nil
|
||||
return revision, entered, err
|
||||
}
|
||||
|
||||
// short is a commit as a person reads it.
|
||||
|
||||
@@ -58,7 +58,9 @@ const (
|
||||
// **Remade when either end's sealing key changes.** A node that rejoined generated a new key and
|
||||
// can no longer open what was sealed to the old one, so keeping the blob would deliver something
|
||||
// unreadable for ever. The new secret reaches both ends in the same push, which is the only
|
||||
// moment they can be changed together.
|
||||
// moment they can be changed together. A remade value is the mesh's, whatever the row held: its
|
||||
// origin says made and it carries no stamp, so a value a provider's software issued is sealed to it
|
||||
// again at the provider's next ask (novox/hq ADR 0305 §5) rather than read as still held.
|
||||
func (i *Inventory) SecretFor(ctx context.Context, name, consumer, consumerModule, provider, local string) (
|
||||
Secret, error) {
|
||||
consumerKey, err := i.SealingKeyOf(ctx, consumer)
|
||||
@@ -121,7 +123,8 @@ func (i *Inventory) SecretFor(ctx context.Context, name, consumer, consumerModul
|
||||
for_consumer = excluded.for_consumer, for_provider = excluded.for_provider,
|
||||
consumer_key = excluded.consumer_key, provider_key = excluded.provider_key,
|
||||
created_at = now(),
|
||||
operator_sealed = excluded.operator_sealed, operator_key = excluded.operator_key`,
|
||||
operator_sealed = excluded.operator_sealed, operator_key = excluded.operator_key,
|
||||
origin = 'made', generation = null`,
|
||||
name, consumerNode.ID, consumerModule, providerNode.ID,
|
||||
made.ForConsumer, made.ForProvider, made.ConsumerKey, made.ProviderKey, forOperator, operatorKey, local)
|
||||
if err != nil {
|
||||
@@ -202,7 +205,7 @@ func (i *Inventory) AcceptSecretForPair(ctx context.Context, name, consumer, con
|
||||
on conflict (name, local, consumer, consumer_module, provider) do update set
|
||||
for_consumer = excluded.for_consumer, for_provider = excluded.for_provider,
|
||||
consumer_key = excluded.consumer_key, provider_key = excluded.provider_key,
|
||||
created_at = now(), origin = excluded.origin,
|
||||
created_at = now(), origin = excluded.origin, generation = null,
|
||||
operator_sealed = excluded.operator_sealed, operator_key = excluded.operator_key`,
|
||||
name, consumerNode.ID, consumerModule, providerNode.ID,
|
||||
sealed.ForConsumer, sealed.ForProvider, sealed.ConsumerKey, sealed.ProviderKey,
|
||||
|
||||
@@ -100,6 +100,23 @@ type Send struct {
|
||||
CounterRaisedTo int64
|
||||
// Recorded is false for a refusal of a sequence the mesh has no send for: its sender is unknown.
|
||||
Recorded bool
|
||||
// Answer is what the machine last reported of this send; filled only where a last send is read back
|
||||
// (LastSend, LastSends), the zero answer elsewhere.
|
||||
Answer SendAnswer
|
||||
}
|
||||
|
||||
// SendAnswer is what a machine reported of one send, read from its last report when that report is about the
|
||||
// send's declaration (novox/hq issue 485): whether it applied it, failed part of it or refused it, and why.
|
||||
// The zero answer is a machine that has not reported on this send — yet, or since a later report replaced it.
|
||||
type SendAnswer struct {
|
||||
// Outcome is applied, failed or refused, as the machine reported it; empty when it has not reported on it.
|
||||
Outcome string
|
||||
// Refused is the machine's own words when it refused the declaration whole.
|
||||
Refused string
|
||||
// Failed is how many of its resources failed, when some did.
|
||||
Failed int
|
||||
// At is when the machine reported it; nil when it has not.
|
||||
At *time.Time
|
||||
}
|
||||
|
||||
// unknownSender is the sender of a refused sequence the mesh has no record of sending.
|
||||
@@ -209,20 +226,56 @@ func scanSends(rows pgx.Rows) ([]Send, error) {
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// LastSend is the last declaration a machine was sent, by its name; false when none was recorded. A refusal
|
||||
// of a sequence the mesh has no send for is not a send, and is not it.
|
||||
// LastSend is the last declaration a machine was sent, by its name, with what the machine answered of it; false
|
||||
// when none was recorded. A refusal of a sequence the mesh has no send for is not a send, and is not it.
|
||||
func (i *Inventory) LastSend(ctx context.Context, name string) (Send, bool, error) {
|
||||
rows, err := i.store.Pool().Query(ctx, `select `+sendColumns+`
|
||||
from declaration_send s join node n on n.id = s.node
|
||||
where n.name = $1 and s.recorded order by s.id desc limit 1`, name)
|
||||
sends, err := i.lastSends(ctx, `n.name = $1`, name)
|
||||
if err != nil {
|
||||
return Send{}, false, err
|
||||
}
|
||||
sends, err := scanSends(rows)
|
||||
if err != nil || len(sends) == 0 {
|
||||
return Send{}, false, err
|
||||
s, found := sends[name]
|
||||
return s, found, nil
|
||||
}
|
||||
|
||||
// LastSends is every machine's last send, by machine name, each with what the machine answered of it: what
|
||||
// `status` says per machine (novox/hq issue 485). A machine never sent anything has no entry.
|
||||
func (i *Inventory) LastSends(ctx context.Context) (map[string]Send, error) {
|
||||
return i.lastSends(ctx, `true`)
|
||||
}
|
||||
|
||||
// lastSends reads the newest recorded send of each machine matching where, beside the machine's last report
|
||||
// when that report is about the send (novox/hq issue 485). It names the send's declaration (its digest), and:
|
||||
// - from an ordered node-engine, one that reports the epoch and sequence it acted on, those are the send's
|
||||
// own. A periodic report about an identical earlier send names the same digest and may arrive after a later
|
||||
// send the machine never received; only the sequence tells them apart. It also holds for a fast machine
|
||||
// whose report lands before the send's row is stamped.
|
||||
// - from an engine that reports no order, the report came after the send — both times the store's own — so
|
||||
// a machine sent again a declaration it once applied does not read as having applied the new send.
|
||||
func (i *Inventory) lastSends(ctx context.Context, where string, args ...any) (map[string]Send, error) {
|
||||
rows, err := i.store.Pool().Query(ctx, `select distinct on (n.name) `+sendColumns+`,
|
||||
coalesce(r.outcome, ''), coalesce(r.refused, ''), case when jsonb_typeof(r.failed) = 'array' then jsonb_array_length(r.failed) else 0 end, r.at
|
||||
from declaration_send s join node n on n.id = s.node
|
||||
left join node_report r on r.node = s.node and r.declared <> '' and r.declared = s.digest
|
||||
and case when r.reported_sequence is not null
|
||||
then r.reported_sequence = s.sequence and coalesce(r.reported_epoch, 0) = coalesce(s.epoch, 0)
|
||||
else r.at >= s.sent_at end
|
||||
where s.recorded and `+where+`
|
||||
order by n.name, s.id desc`, args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return sends[0], true, nil
|
||||
defer rows.Close()
|
||||
out := map[string]Send{}
|
||||
for rows.Next() {
|
||||
var s Send
|
||||
if err := rows.Scan(&s.Node, &s.NodeName, &s.Sequence, &s.Epoch, &s.Sender, &s.Generation, &s.Digest,
|
||||
&s.Modules, &s.SentAt, &s.RefusedAt, &s.RefusedApplied, &s.CounterRaisedTo, &s.Recorded,
|
||||
&s.Answer.Outcome, &s.Answer.Refused, &s.Answer.Failed, &s.Answer.At); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out[s.NodeName] = s
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// RefusedSend keeps a machine's refusal of the send of a sequence for the generation it came from, and
|
||||
|
||||
@@ -205,3 +205,144 @@ func TestWhetherAMachineReadsAGenerationIsItsLatestWord(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// **The last send says what the machine answered of it** (novox/hq issue 485): applied, refused with its
|
||||
// words, or not answered yet — read from the machine's report only when that report is about the send's
|
||||
// declaration and came after it, so a report about an earlier send is never this one's answer. And every
|
||||
// machine's last send is read at once, for status, by name; a machine never sent anything has none.
|
||||
func TestTheLastSendSaysWhatTheMachineAnswered(t *testing.T) {
|
||||
inv, node := aNodeWithModules(t)
|
||||
ctx := t.Context()
|
||||
record, err := inv.NodeByName(ctx, node)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := inv.AddNode(ctx, "laptop"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
send := func(sequence int64, digest string) {
|
||||
t.Helper()
|
||||
if err := inv.RecordSentWith(ctx, record.ID, digest, nil, 57, 40, Send{Sequence: sequence, Epoch: 57,
|
||||
Sender: "the controller's daemon (pid 7 on anchor)", Generation: 40, Digest: digest,
|
||||
Modules: []string{"docker", "sudo"}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
report := func(d Doing) {
|
||||
t.Helper()
|
||||
if _, err := inv.RecordDoing(ctx, record.ID, d); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
answer := func() SendAnswer {
|
||||
t.Helper()
|
||||
last, found, err := inv.LastSend(ctx, node)
|
||||
if err != nil || !found {
|
||||
t.Fatalf("the last send is not read back: %v, %v", found, err)
|
||||
}
|
||||
return last.Answer
|
||||
}
|
||||
|
||||
send(11, "d11")
|
||||
if a := answer(); a.Outcome != "" || a.At != nil {
|
||||
t.Fatalf("a send not reported on reads as answered: %+v", a)
|
||||
}
|
||||
report(Doing{Outcome: OutcomeApplied, Declared: "d11", Applied: 3})
|
||||
if a := answer(); a.Outcome != OutcomeApplied || a.At == nil {
|
||||
t.Fatalf("a send the machine reported applying reads %+v", a)
|
||||
}
|
||||
|
||||
send(12, "d12")
|
||||
if a := answer(); a.Outcome != "" {
|
||||
t.Fatalf("the report about the send before is taken for this one's answer: %+v", a)
|
||||
}
|
||||
report(Doing{Outcome: OutcomeRefused, Declared: "d12", Refused: "unknown field \"generation\"\nand more"})
|
||||
if a := answer(); a.Outcome != OutcomeRefused || a.Refused != "unknown field \"generation\"\nand more" {
|
||||
t.Fatalf("a send the machine refused reads %+v", a)
|
||||
}
|
||||
|
||||
// Sent again a declaration it reported applying before the send: not answered until it reports again.
|
||||
report(Doing{Outcome: OutcomeApplied, Declared: "d12", Applied: 3})
|
||||
send(13, "d12")
|
||||
if a := answer(); a.Outcome != "" {
|
||||
t.Fatalf("a report from before the send is taken for its answer: %+v", a)
|
||||
}
|
||||
|
||||
all, err := inv.LastSends(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(all) != 1 || all[node].Sequence != 13 || all[node].Epoch != 57 || all[node].Generation != 40 ||
|
||||
!slices.Equal(all[node].Modules, []string{"docker", "sudo"}) {
|
||||
t.Fatalf("every machine's last send reads %+v", all)
|
||||
}
|
||||
if _, found := all["laptop"]; found {
|
||||
t.Error("a machine never sent anything has a last send")
|
||||
}
|
||||
}
|
||||
|
||||
// **A report about another declaration, or another send of the same one, is not the last send's answer**
|
||||
// (novox/hq issue 485, review of mesh-controller #234). A report naming an earlier send's digest that arrives
|
||||
// after a later send is not the later send's answer. And from an ordered node-engine, a periodic report about
|
||||
// an identical earlier send — the same digest, arriving after the later send — is not its answer either: only
|
||||
// the report's epoch and sequence say which send it is about.
|
||||
func TestAReportIsTheLastSendsAnswerOnlyWhenItIsAboutThatSend(t *testing.T) {
|
||||
inv, node := aNodeWithModules(t)
|
||||
ctx := t.Context()
|
||||
record, err := inv.NodeByName(ctx, node)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
send := func(sequence int64, digest string) {
|
||||
t.Helper()
|
||||
if err := inv.RecordSentWith(ctx, record.ID, digest, nil, 57, 40, Send{Sequence: sequence, Epoch: 57,
|
||||
Sender: "the controller's daemon (pid 7 on anchor)", Generation: 40, Digest: digest}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
ordered := func(epoch, sequence int64, digest string) {
|
||||
t.Helper()
|
||||
if _, err := inv.RecordOrderedDoing(ctx, record.ID, Doing{Outcome: OutcomeApplied, Declared: digest, Applied: 1},
|
||||
ReportOrder{Epoch: epoch, Sequence: sequence}, func(ReportOrder) bool { return false }); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
answer := func() string {
|
||||
t.Helper()
|
||||
last, found, err := inv.LastSend(ctx, node)
|
||||
if err != nil || !found {
|
||||
t.Fatalf("the last send is not read back: %v, %v", found, err)
|
||||
}
|
||||
return last.Answer.Outcome
|
||||
}
|
||||
|
||||
// An engine that reports no order: the digest decides.
|
||||
send(11, "d11")
|
||||
send(12, "d12")
|
||||
if _, err := inv.RecordDoing(ctx, record.ID, Doing{Outcome: OutcomeApplied, Declared: "d11"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if a := answer(); a != "" {
|
||||
t.Fatalf("a report about d11, after d12 was sent, reads as d12's answer: %q", a)
|
||||
}
|
||||
|
||||
// An ordered engine: the epoch and sequence decide, whatever the digest and the time.
|
||||
send(20, "d20")
|
||||
ordered(57, 20, "d20")
|
||||
if a := answer(); a != OutcomeApplied {
|
||||
t.Fatalf("an ordered report about sequence 20 is not its answer: %q", a)
|
||||
}
|
||||
send(21, "d20")
|
||||
ordered(57, 20, "d20")
|
||||
if a := answer(); a != "" {
|
||||
t.Fatalf("a periodic report about sequence 20, after 21 was sent with the same digest, reads as 21's answer: %q", a)
|
||||
}
|
||||
ordered(56, 21, "d20")
|
||||
if a := answer(); a != "" {
|
||||
t.Fatalf("a report about sequence 21 of another epoch reads as this send's answer: %q", a)
|
||||
}
|
||||
ordered(57, 21, "d20")
|
||||
if a := answer(); a != OutcomeApplied {
|
||||
t.Fatalf("an ordered report about sequence 21 is not its answer: %q", a)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user