Compare commits
19
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
516af253c6 | ||
|
|
12752911b6 | ||
|
|
087e9266d1 | ||
|
|
a3b4e0601d | ||
|
|
da364c2730 | ||
|
|
98ffcb667f | ||
|
|
2aa291f3b3 | ||
|
|
ceaa616d01 | ||
|
|
fa5b5e1c11 | ||
|
|
6ef78580ce | ||
|
|
0e0c775bfa | ||
|
|
61c8b56107 | ||
|
|
c24139a0fd | ||
|
|
d8d6d196a4 | ||
|
|
f10fbda1f6 | ||
|
|
09aa063a13 | ||
|
|
5be1cfaed1 | ||
|
|
35829d5a9a | ||
|
|
57939d6dd4 |
@@ -232,6 +232,8 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T
|
||||
fmt.Print(nothingReadsIt(repository, m))
|
||||
return nil
|
||||
}
|
||||
// **A module the merge adds is asked for now** (novox/hq issue 1617), not when the batch is cut.
|
||||
askNewModulesAtHearing(ctx, inv, m, entries, read, facts.from)
|
||||
// **Its delivery group's word, before the plans are held** (ADR 0306 decision 2): a slow answer holds up no walk.
|
||||
own := ownPath(touches)
|
||||
waits, holding, err := groupWaitHolding(ctx, inv, m, deliverySeatHeld(entries))
|
||||
|
||||
@@ -32,6 +32,13 @@ func assertBusObjects(ctx context.Context, inv *inventory.Inventory, r broker.Ra
|
||||
if err := broker.Raise(r, names); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// And each node's bucket for the files of its declaration that travel by digest (novox/hq ADR 0486 §4), so a node finds its own
|
||||
// whenever the bus is up.
|
||||
if files, ok := r.(interface{ EnsureDeclarationFiles(nodes []string) error }); ok {
|
||||
if err := files.EnsureDeclarationFiles(names); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
// The work queues of the mesh's own roles (novox/hq ADR 0121). The queue before the holder,
|
||||
// deliberately: work queues until somebody arrives to do it, so assigning a build machine a week
|
||||
// after something started asking for builds flushes the backlog instead of having lost it.
|
||||
|
||||
@@ -200,6 +200,7 @@ func composeQuestion(ctx context.Context, open *stores, node string, mode compos
|
||||
}
|
||||
sayFound(ctx, d)
|
||||
out := sendableOf(d)
|
||||
out.fileReferences = readsFileReferences(in.Input, node)
|
||||
if !withOrder {
|
||||
out.Sequence, out.Epoch, out.Generation = 0, 0, 0
|
||||
}
|
||||
@@ -281,6 +282,7 @@ func composeSendValidated(ctx context.Context, open *stores, node string) (senda
|
||||
if len(left) == 0 && len(in.wouldMake) == 0 {
|
||||
sayFound(ctx, d)
|
||||
out := sendableOf(d)
|
||||
out.fileReferences = readsFileReferences(in.Input, node)
|
||||
// The sender stamps the order it allots (order.stamp); a composition for a send carries none of its own.
|
||||
out.Sequence, out.Epoch, out.Generation = 0, 0, 0
|
||||
here, _ := nodeOf(in.Input, node)
|
||||
@@ -366,7 +368,7 @@ func sayFound(ctx context.Context, d compose.Declared) {
|
||||
|
||||
// sendableOf is a composed declaration as the controller's sends know it.
|
||||
func sendableOf(d compose.Declared) sendable {
|
||||
out := sendable{Resources: d.Resources, Sequence: d.Sequence, Epoch: d.Epoch, Generation: d.Generation,
|
||||
out := sendable{Resources: compactKept(d.Resources), Sequence: d.Sequence, Epoch: d.Epoch, Generation: d.Generation,
|
||||
ReadsModules: d.ReadsModules, moduleOf: d.ModuleOf, buildIDs: d.BuildIDs, modules: d.Modules,
|
||||
localNetwork: d.LocalNetwork, Received: d.Received, Mesh: d.Mesh, BusUsers: d.BusUsers,
|
||||
LeftOut: d.LeftOut, leftOutWhy: d.LeftOutWhy, withheld: d.Withheld, unbound: d.Unbound, foreseen: d.Foreseen,
|
||||
|
||||
@@ -0,0 +1,65 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"io"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
)
|
||||
|
||||
// Issue 1638 (novox/hq ADR 0480 clause 7), taken in with mesh-sdk's container `groups`: the controller sends a container's
|
||||
// groups only to a node-engine that reports the capability container-groups, and holds the module on any other, naming it.
|
||||
|
||||
const n8nWithAGroup = `{"module":"n8n","version":"1",
|
||||
"config":{"media-gid":{"type":"number","minimum":1,"required":true,"description":"the group that owns the media library"}},
|
||||
"resources":[{"id":"server","type":"container","name":"n8n",
|
||||
"image":"n8nio/n8n@sha256:4846eb2f4b874ab04cde7fc1e249d2ddaec66e9aea64439beb2972cfea88e3c0",
|
||||
"groups":["${config:media-gid}"]}]}`
|
||||
|
||||
// printedBy runs reportUnhostable and returns what it printed.
|
||||
func printedBy(t *testing.T, node string, plan catalogue.Resolution) string {
|
||||
t.Helper()
|
||||
stdout := os.Stdout
|
||||
r, w, err := os.Pipe()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
os.Stdout = w
|
||||
reportUnhostable(node, plan)
|
||||
w.Close()
|
||||
os.Stdout = stdout
|
||||
out, _ := io.ReadAll(r)
|
||||
return string(out)
|
||||
}
|
||||
|
||||
// 1638/AC1: a node that reports the capability gets the module; one that does not has it held, the capability named.
|
||||
func TestAModuleWithGroupsIsHeldOnANodeThatDoesNotReportTheCapability(t *testing.T) {
|
||||
m, err := catalogue.ParseManifest([]byte(n8nWithAGroup))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
shelf := catalogue.Shelf{"n8n": m}
|
||||
reads := catalogue.Node{Name: "ace", Capabilities: map[string]bool{"container-runtime": true, "container-groups": true}}
|
||||
old := catalogue.Node{Name: "g14", Capabilities: map[string]bool{"container-runtime": true}}
|
||||
|
||||
got, err := catalogue.Resolve(shelf, []string{"n8n"}, reads, catalogue.World{})
|
||||
if err != nil || len(got.Unhostable) != 0 || len(got.Modules) != 1 {
|
||||
t.Fatalf("a node that reads groups: %v, held %+v", err, got.Unhostable)
|
||||
}
|
||||
if out := printedBy(t, "ace", got); out != "" {
|
||||
t.Errorf("a node that reads groups was told: %q", out)
|
||||
}
|
||||
|
||||
got, err = catalogue.Resolve(shelf, []string{"n8n"}, old, catalogue.World{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(got.Modules) != 0 || len(got.Unhostable) != 1 {
|
||||
t.Fatalf("n8n was sent to a node that does not report container-groups: hosted %d, held %+v", len(got.Modules), got.Unhostable)
|
||||
}
|
||||
if out := printedBy(t, "g14", got); !strings.Contains(out, "n8n") || !strings.Contains(out, "container-groups") {
|
||||
t.Errorf("the hold does not name the module and the capability: %q", out)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,110 @@
|
||||
//go:build integration
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
sdkdecl "git.novox.be/novox/mesh-sdk/go/declaration"
|
||||
"git.novox.be/novox/mesh-sdk/go/overlay"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/compose"
|
||||
)
|
||||
|
||||
// 1696/AC2, ADR 0486 §5: a node is sent a file by digest only when its profile reports the capability. A node whose profile lacks
|
||||
// it, and a node the mesh has no profile of, are sent every content inline; a node that reports it is sent the reference. The test
|
||||
// goes through composeQuestion and composeSend, the two places readsFileReferences is read (compose_sdk.go).
|
||||
func TestAFileTravelsByDigestOnlyToANodeThatReportsTheCapability(t *testing.T) {
|
||||
open := aMesh(t) // anchor and laptop report no file-references
|
||||
ctx := t.Context()
|
||||
big := strings.Repeat("0123456789abcdef", 8<<10) // 128 KiB
|
||||
register(t, open, catalogue.Manifest{Module: "bigfile", Version: "1",
|
||||
Resources: []map[string]any{{"id": "blob", "type": "file", "path": "/etc/bigfile/blob", "mode": "0644", "content": big}}})
|
||||
|
||||
// reader reports the capability; bare has never reported a profile at all.
|
||||
for i, n := range []struct {
|
||||
name string
|
||||
profile map[string]any
|
||||
}{
|
||||
{"reader", map[string]any{"capabilities": []any{
|
||||
map[string]any{"name": "container-runtime", "present": true},
|
||||
map[string]any{"name": "systemd", "present": true},
|
||||
map[string]any{"name": sdkdecl.FileReferencesCapability, "present": true}}}},
|
||||
{"bare", nil},
|
||||
} {
|
||||
record, err := open.inventory.AddNode(ctx, n.name)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := open.inventory.SetPlace(ctx, n.name, n.name+".example:51820", "here", false, fmt.Sprintf("10.77.0.%d", i+10)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n.profile != nil {
|
||||
if err := open.inventory.RecordProfile(ctx, record.ID, n.profile); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if err := open.inventory.RecordSealingKey(ctx, record.ID, aPublicKey(t)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := open.inventory.RecordOverlayKey(ctx, record.ID, aPublicKey(t)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := open.inventory.Assign(ctx, n.name, overlay.Name); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
for _, node := range []string{"laptop", "reader", "bare"} {
|
||||
if _, err := assign(ctx, open, node, "bigfile"); err != nil {
|
||||
t.Fatalf("assign %s: %v", node, err)
|
||||
}
|
||||
}
|
||||
|
||||
byDigest := map[string]bool{"laptop": false, "bare": false, "reader": true}
|
||||
for node, want := range byDigest {
|
||||
asked, _, err := composeQuestion(ctx, open, node, compose.Reading, false)
|
||||
if err != nil {
|
||||
t.Fatalf("%s: %v", node, err)
|
||||
}
|
||||
sent, err := composeSend(ctx, open, node)
|
||||
if err != nil {
|
||||
t.Fatalf("%s: %v", node, err)
|
||||
}
|
||||
for what, s := range map[string]sendable{"question": asked, "send": sent} {
|
||||
if s.fileReferences != want {
|
||||
t.Fatalf("%s, %s: fileReferences %v, want %v", node, what, s.fileReferences, want)
|
||||
}
|
||||
body, err := s.Body()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var got struct {
|
||||
Resources []map[string]any `json:"resources"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &got); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var found bool
|
||||
for _, r := range got.Resources {
|
||||
if r["path"] != "/etc/bigfile/blob" {
|
||||
continue
|
||||
}
|
||||
found = true
|
||||
inline := r["content"] == big
|
||||
if want && (inline || r["digest"] == nil || r["size"] == nil) {
|
||||
t.Fatalf("%s, %s: a node that reads references is sent the content inline", node, what)
|
||||
}
|
||||
if !want && (!inline || r["digest"] != nil) {
|
||||
t.Fatalf("%s, %s: a node that does not report the capability is sent a reference", node, what)
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Fatalf("%s, %s: the file is not in the declaration", node, what)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,253 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"sync"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/compose"
|
||||
sdkdecl "git.novox.be/novox/mesh-sdk/go/declaration"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
"github.com/novox/mesh-controller/internal/conditions"
|
||||
)
|
||||
|
||||
// **A declaration's large files travel by their digest to a node that can read one** (novox/hq ADR 0486 §1, §5, issue 1696).
|
||||
//
|
||||
// The declaration to a node that reports sdkdecl.FileReferencesCapability carries, in place of the content of every
|
||||
// file resource of more than sdkdecl.FileReferenceThreshold bytes that may travel so (File.TravelsByDigest: plain content or
|
||||
// bytes, not sealed, not filled with secrets, not written into another file), its sha256 digest and size. The content is put
|
||||
// into the bucket of the bus under the hex of the digest before the declaration is sent. A node that does not report the
|
||||
// capability is sent every content inline, as before: an older node-engine refuses a field it does not know, whole.
|
||||
|
||||
// readsFileReferences says the node reports the capability of reading a file given by digest (ADR 0486 §5). A node the mesh has
|
||||
// no profile of reads none.
|
||||
func readsFileReferences(in compose.Input, node string) bool {
|
||||
n, ok := nodeOf(in, node)
|
||||
return ok && n.Capabilities[sdkdecl.FileReferencesCapability]
|
||||
}
|
||||
|
||||
// referencing is the resources as a node that reads references is sent them, and the content of each file that went by
|
||||
// digest, by the name of its object (the hex of its digest). The resources it is given are not changed. A file that does not
|
||||
// travel by digest is passed as it is.
|
||||
func referencing(resources []map[string]any) ([]map[string]any, map[string][]byte, error) {
|
||||
out := make([]map[string]any, len(resources))
|
||||
var objects map[string][]byte
|
||||
for i, r := range resources {
|
||||
out[i] = r
|
||||
if kind, _ := r["type"].(string); kind != string(sdkdecl.TypeFile) {
|
||||
continue
|
||||
}
|
||||
// Cheap first: only a file with a content or bytes of the size is worth reading whole.
|
||||
content, _ := r["content"].(string)
|
||||
bytesField, _ := r["bytes"].(string)
|
||||
if len(content) <= sdkdecl.FileReferenceThreshold && len(bytesField) <= sdkdecl.FileReferenceThreshold {
|
||||
continue
|
||||
}
|
||||
raw, err := json.Marshal(r)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("resource %v: %w", r["id"], err)
|
||||
}
|
||||
var f sdkdecl.File
|
||||
if err := json.Unmarshal(raw, &f); err != nil {
|
||||
return nil, nil, fmt.Errorf("resource %v is not a file the controller can read: %w", r["id"], err)
|
||||
}
|
||||
if !f.TravelsByDigest() {
|
||||
continue
|
||||
}
|
||||
ref, body, err := f.AsReference()
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
name, err := sdkdecl.ObjectName(ref.Digest)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
with := make(map[string]any, len(r)+2)
|
||||
for k, v := range r {
|
||||
with[k] = v
|
||||
}
|
||||
delete(with, "content")
|
||||
delete(with, "bytes")
|
||||
with["digest"], with["size"] = ref.Digest, ref.Size
|
||||
out[i] = with
|
||||
if objects == nil {
|
||||
objects = map[string][]byte{}
|
||||
}
|
||||
objects[name] = body
|
||||
}
|
||||
return out, objects, nil
|
||||
}
|
||||
|
||||
// namedObjects are the objects of the bucket a body names: the object name of every file resource given by reference,
|
||||
// with its size. It reads a declaration as the node does, so it also reads the latest one the bus holds.
|
||||
func namedObjects(body []byte) (map[string]int64, error) {
|
||||
var envelope struct {
|
||||
Resources []struct {
|
||||
Type string `json:"type"`
|
||||
Digest string `json:"digest"`
|
||||
Size int64 `json:"size"`
|
||||
} `json:"resources"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &envelope); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var out map[string]int64
|
||||
for _, r := range envelope.Resources {
|
||||
if r.Type != string(sdkdecl.TypeFile) || r.Digest == "" {
|
||||
continue
|
||||
}
|
||||
name, err := sdkdecl.ObjectName(r.Digest)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if out == nil {
|
||||
out = map[string]int64{}
|
||||
}
|
||||
out[name] = r.Size
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// objectStore is what putting the objects needs of a node's bucket.
|
||||
type objectStore interface {
|
||||
// Has says whether an object of that name is there, complete, and whether it carries the sweep's mark of being named by no
|
||||
// node's latest declaration.
|
||||
Has(ctx context.Context, name string) (have, marked bool, err error)
|
||||
// Put writes the object under that name, replacing one of the same name (and so erasing its mark).
|
||||
Put(ctx context.Context, name string, content []byte) error
|
||||
}
|
||||
|
||||
// putObjects makes sure every object of the declaration is in the bucket before the declaration that names it is sent: a node
|
||||
// that reads the declaration finds each object. An object already there (the same content, the same name) is not written again,
|
||||
// **unless the sweep marked it unnamed**: then it is written again, so the mark that would have it removed is gone with the
|
||||
// old object (issue 1696, review: a mark read just before a send must not take an object the send relies on). Refuses on the
|
||||
// first it cannot put, naming it, so the declaration is not sent.
|
||||
func putObjects(ctx context.Context, store objectStore, objects map[string][]byte) error {
|
||||
names := make([]string, 0, len(objects))
|
||||
for name := range objects {
|
||||
names = append(names, name)
|
||||
}
|
||||
sort.Strings(names)
|
||||
for _, name := range names {
|
||||
have, marked, err := store.Has(ctx, name)
|
||||
if err != nil {
|
||||
return fmt.Errorf("whether the object %s is in the bucket: %w", name[:12], err)
|
||||
}
|
||||
if have && !marked {
|
||||
continue
|
||||
}
|
||||
if err := store.Put(ctx, name, objects[name]); err != nil {
|
||||
return fmt.Errorf("putting the object %s (%d bytes) in the bucket: %w", name[:12], len(objects[name]), err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// busObjects is a node's bucket on the controller's own connection.
|
||||
type busObjects struct{ store jetstream.ObjectStore }
|
||||
|
||||
// openBucket opens the bucket of a node's files, making it first if it is not there (it is asserted for every node at the
|
||||
// controller's start; a node adopted since is covered here).
|
||||
func openBucket(ctx context.Context, conn *nats.Conn, node string) (busObjects, error) {
|
||||
js, err := jetstream.New(conn)
|
||||
if err != nil {
|
||||
return busObjects{}, err
|
||||
}
|
||||
bucket, err := broker.DeclarationFilesBucketOf(node)
|
||||
if err != nil {
|
||||
return busObjects{}, fmt.Errorf("the object bucket of node %q: %w", node, err)
|
||||
}
|
||||
store, err := js.ObjectStore(ctx, bucket)
|
||||
if errors.Is(err, jetstream.ErrBucketNotFound) {
|
||||
if err = broker.EnsureDeclarationFiles(ctx, js, node); err == nil {
|
||||
store, err = js.ObjectStore(ctx, bucket)
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
return busObjects{}, fmt.Errorf("the object bucket %s: %w", bucket, err)
|
||||
}
|
||||
return busObjects{store}, nil
|
||||
}
|
||||
|
||||
func (b busObjects) Has(ctx context.Context, name string) (bool, bool, error) {
|
||||
info, err := b.store.GetInfo(ctx, name)
|
||||
switch {
|
||||
case errors.Is(err, jetstream.ErrObjectNotFound):
|
||||
return false, false, nil
|
||||
case err != nil:
|
||||
return false, false, err
|
||||
}
|
||||
_, marked := info.Metadata[declarationUnnamed]
|
||||
return !info.Deleted, marked, nil
|
||||
}
|
||||
|
||||
func (b busObjects) Put(ctx context.Context, name string, content []byte) error {
|
||||
_, err := b.store.PutBytes(ctx, name, content)
|
||||
return err
|
||||
}
|
||||
|
||||
// filesLocks hold a node's bucket for one act at a time in this process: a send puts its objects and publishes the declaration that
|
||||
// names them under the node's lock, and the sweep judges the node's bucket under the same one, so the sweep never removes an
|
||||
// object between a send's put and its declaration. Another process (a person's push) is covered by the put erasing a mark and by
|
||||
// the sweep looking at an object again just before it removes it.
|
||||
var filesLocks sync.Map
|
||||
|
||||
func lockFilesOf(node string) func() {
|
||||
l, _ := filesLocks.LoadOrStore(node, &sync.Mutex{})
|
||||
mu := l.(*sync.Mutex)
|
||||
mu.Lock()
|
||||
return mu.Unlock
|
||||
}
|
||||
|
||||
// kindFilesNotPut is a node whose files that travel by digest could not be put in its bucket, send after send.
|
||||
const kindFilesNotPut = "declaration-files-not-put"
|
||||
|
||||
// filesNotPutAfter is how many sends in a row may fail to put before the condition is raised: the first send to the machine holding
|
||||
// the bus fails for want of the grant that very send delivers, and says so, and is not yet a fault.
|
||||
const filesNotPutAfter = 2
|
||||
|
||||
var (
|
||||
filesFailMu sync.Mutex
|
||||
filesFailed = map[string]int{}
|
||||
)
|
||||
|
||||
func filesNotPutObservation(node, summary string) conditions.Observation {
|
||||
return conditions.Observation{Scope: conditions.ScopeMachine, ID: node, Kind: kindFilesNotPut, Token: kindFilesNotPut,
|
||||
Machine: node, Severity: conditions.Warning, Source: "sends", Summary: summary}
|
||||
}
|
||||
|
||||
// notePut records how putting a send's objects went: a failure counts, the second in a row raises the condition, a success
|
||||
// clears the count and the condition.
|
||||
func notePut(ctx context.Context, node string, put error) {
|
||||
filesFailMu.Lock()
|
||||
if put == nil {
|
||||
delete(filesFailed, node)
|
||||
} else {
|
||||
filesFailed[node]++
|
||||
}
|
||||
n := filesFailed[node]
|
||||
filesFailMu.Unlock()
|
||||
keeper := conditionsFrom
|
||||
if keeper == nil {
|
||||
return
|
||||
}
|
||||
key := filesNotPutObservation(node, "").Key()
|
||||
switch {
|
||||
case put == nil:
|
||||
if _, err := keeper.Clear(ctx, key, node+" was sent its files by digest"); err != nil {
|
||||
fmt.Printf("the condition %s could not be cleared: %v\n", key, err)
|
||||
}
|
||||
case n >= filesNotPutAfter:
|
||||
summary := fmt.Sprintf("%s's files that travel by digest could not be put in its bucket %d sends in a row, so it is sent them "+
|
||||
"inline: %v", node, n, put)
|
||||
if _, err := keeper.Observe(ctx, filesNotPutObservation(node, summary)); err != nil {
|
||||
fmt.Printf("the condition %s could not be raised: %v\n", key, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,246 @@
|
||||
//go:build integration
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
sdkdecl "git.novox.be/novox/mesh-sdk/go/declaration"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
"github.com/novox/mesh-controller/internal/link"
|
||||
"github.com/novox/mesh-controller/internal/testbus"
|
||||
)
|
||||
|
||||
type sigOf struct{}
|
||||
|
||||
func (sigOf) Sign(context.Context, []byte) ([]byte, error) { return make([]byte, 64), nil }
|
||||
|
||||
// 1696/AC7 (ADR 0486 §1): on a real bus with its default largest message of 1 MiB, a declaration whose files add up to more than
|
||||
// 3 MiB is sent as references, its objects are in the bucket, each fetched back whole and verified against the digest and size the
|
||||
// declaration names; the same declaration inline is refused by the controller, naming size and limit. A changed byte in an object
|
||||
// fails the verification.
|
||||
func TestADeclarationOfMoreThanThreeMiBIsSentByReferenceOnARealBus(t *testing.T) { // 1696/AC7
|
||||
s := testbus.Start(t)
|
||||
conn, err := nats.Connect(s.ClientURL())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer conn.Close()
|
||||
nc, err := conn.JetStream()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := nc.AddStream(&nats.StreamConfig{Name: "NODES", Subjects: []string{"mesh.node.*.declare"},
|
||||
MaxMsgsPerSubject: 1}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
||||
defer cancel()
|
||||
api, err := jetstream.New(conn)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := broker.EnsureDeclarationFiles(ctx, api, "novox"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// Run twice: an assertion of an existing bucket is no error and no change.
|
||||
if err := broker.EnsureDeclarationFiles(ctx, api, "novox"); err != nil {
|
||||
t.Fatalf("asserting the bucket again: %v", err)
|
||||
}
|
||||
|
||||
var resources []map[string]any
|
||||
for i := 0; i < 3; i++ {
|
||||
resources = append(resources, aFileResource("big-"+string(rune('a'+i)), contentOf("big"+string(rune('a'+i)), 1200<<10)))
|
||||
}
|
||||
resources = append(resources, aFileResource("small", "tiny\n"))
|
||||
inline := sendable{Resources: resources}
|
||||
byDigest := sendable{fileReferences: true, Resources: resources}
|
||||
|
||||
bus := link.OverNATS{Conn: conn, JS: nc}
|
||||
body, err := inline.Body()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := link.Declare(ctx, bus, sigOf{}, "novox", body, 5*time.Second); err == nil ||
|
||||
!strings.Contains(err.Error(), "1048576") || strings.Contains(err.Error(), "maximum payload exceeded") {
|
||||
t.Fatalf("the inline declaration of more than 3 MiB: %v", err)
|
||||
}
|
||||
|
||||
bucket, err := openBucket(ctx, conn, "novox")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
objects, err := byDigest.objects()
|
||||
if err != nil || len(objects) != 3 {
|
||||
t.Fatalf("objects: %d %v", len(objects), err)
|
||||
}
|
||||
if err := putObjects(ctx, bucket, objects); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := putObjects(ctx, bucket, objects); err != nil {
|
||||
t.Fatalf("putting them again: %v", err)
|
||||
}
|
||||
refBody, err := byDigest.Body()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := link.Declare(ctx, bus, sigOf{}, "novox", refBody, 5*time.Second); err != nil {
|
||||
t.Fatalf("the declaration by reference: %v", err)
|
||||
}
|
||||
|
||||
// What the stream holds for the node is the reference form, and it names exactly the three objects.
|
||||
latest, err := latestDeclarations(ctx, conn, []string{"novox", "ace"})
|
||||
if err != nil || len(latest) != 1 {
|
||||
t.Fatalf("the latest declarations: %v %v", len(latest), err)
|
||||
}
|
||||
named, err := namedByLatest("novox", latest["novox"])
|
||||
if err != nil || len(named) != 3 {
|
||||
t.Fatalf("what the latest declaration names: %v %v", named, err)
|
||||
}
|
||||
novoxBucket, err := broker.DeclarationFilesBucketOf("novox")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
store, err := api.ObjectStore(ctx, novoxBucket)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var envelope struct {
|
||||
Resources []sdkdecl.File `json:"resources"`
|
||||
}
|
||||
if err := json.Unmarshal(refBody, &envelope); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, f := range envelope.Resources {
|
||||
if !f.IsReference() {
|
||||
continue
|
||||
}
|
||||
name, _ := sdkdecl.ObjectName(f.Digest)
|
||||
got, err := store.GetBytes(ctx, name)
|
||||
if err != nil {
|
||||
t.Fatalf("the object of %s: %v", f.ID, err)
|
||||
}
|
||||
if err := f.Verify(got); err != nil {
|
||||
t.Errorf("the object does not verify: %v", err)
|
||||
}
|
||||
got[len(got)/2] ^= 1
|
||||
if err := f.Verify(got); err == nil {
|
||||
t.Errorf("a changed byte in the object of %s passed the verification", f.ID)
|
||||
}
|
||||
}
|
||||
|
||||
// The sweep against the real bucket: all named, nothing marked; a fourth object named by none is marked, kept a day less a
|
||||
// minute, and removed at the day.
|
||||
extra := bytes.Repeat([]byte("x"), 100)
|
||||
extraName := strings.Repeat("e", 64)
|
||||
if err := putObjects(ctx, bucket, map[string][]byte{extraName: extra}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b := busSweptBucket{store}
|
||||
t0 := time.Now()
|
||||
if r, err := sweepDeclarationFiles(ctx, b, named, t0); err != nil || r.Marked != 1 || r.Removed != 0 || r.Kept != 3 {
|
||||
t.Fatalf("first sweep: %+v %v", r, err)
|
||||
}
|
||||
if r, err := sweepDeclarationFiles(ctx, b, named, t0.Add(24*time.Hour-time.Minute)); err != nil || r.Removed != 0 {
|
||||
t.Fatalf("a minute short of a day: %+v %v", r, err)
|
||||
}
|
||||
if r, err := sweepDeclarationFiles(ctx, b, named, t0.Add(24*time.Hour+time.Second)); err != nil || r.Removed != 1 {
|
||||
t.Fatalf("a day after: %+v %v", r, err)
|
||||
}
|
||||
if infos, err := store.List(ctx); err != nil || len(infos) != 3 {
|
||||
t.Fatalf("the bucket holds %d objects (%v), want the 3 the declaration names", len(infos), err)
|
||||
}
|
||||
}
|
||||
|
||||
// 1696/AC8 on a real bucket: the sweep removes an object only if it still carries the mark it judged it by. An object marked
|
||||
// unnamed, then put again by a send (which erases the mark), is kept when the sweep deletes it by the old mark.
|
||||
func TestTheSweepsDeleteLeavesAnObjectASendPutAgainAfterItWasMarked(t *testing.T) { // 1696/AC8
|
||||
s := testbus.Start(t)
|
||||
conn, err := nats.Connect(s.ClientURL())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer conn.Close()
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
||||
defer cancel()
|
||||
bucket, err := openBucket(ctx, conn, "novox")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
name := strings.Repeat("a", 64)
|
||||
content := map[string][]byte{name: bytes.Repeat([]byte("y"), 100)}
|
||||
if err := putObjects(ctx, bucket, content); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b := busSweptBucket{bucket.store}
|
||||
judged := time.Now().Add(-25 * time.Hour).Truncate(time.Second)
|
||||
if err := b.MarkUnnamed(ctx, name, judged); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, marked, err := bucket.Has(ctx, name); err != nil || !marked {
|
||||
t.Fatalf("the object was not marked: %v %v", marked, err)
|
||||
}
|
||||
// The sweep has read the mark; a send now puts the object again, which erases it.
|
||||
if err := putObjects(ctx, bucket, content); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if have, marked, err := bucket.Has(ctx, name); err != nil || !have || marked {
|
||||
t.Fatalf("a send did not put the marked object again: have %v marked %v %v", have, marked, err)
|
||||
}
|
||||
gone, err := b.Delete(ctx, name, judged)
|
||||
if err != nil || gone {
|
||||
t.Fatalf("the sweep deleted by an old mark: gone %v, %v", gone, err)
|
||||
}
|
||||
if have, _, err := bucket.Has(ctx, name); err != nil || !have {
|
||||
t.Fatalf("the object a send relies on is gone: %v %v", have, err)
|
||||
}
|
||||
// And the mark still on it is honoured: marked again and deleted by that mark, it goes.
|
||||
if err := b.MarkUnnamed(ctx, name, judged); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if gone, err := b.Delete(ctx, name, judged); err != nil || !gone {
|
||||
t.Fatalf("an object still carrying the mark it was judged by is kept: gone %v, %v", gone, err)
|
||||
}
|
||||
}
|
||||
|
||||
// 1696/AC9 on a real bus: the sweep of a node's bucket takes the node's lock first, so it waits for a send that has put its objects
|
||||
// and not yet published the declaration that names them.
|
||||
func TestTheSweepOfANodesBucketHoldsTheNodesLock(t *testing.T) { // 1696/AC9
|
||||
s := testbus.Start(t)
|
||||
conn, err := nats.Connect(s.ClientURL())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer conn.Close()
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
||||
defer cancel()
|
||||
const node = "sweep-lock-node"
|
||||
if _, err := openBucket(ctx, conn, node); err != nil { // the bucket exists, or the sweep would return before the lock
|
||||
t.Fatal(err)
|
||||
}
|
||||
release := lockFilesOf(node) // a send is between its put and its publication
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
_, _ = sweepNode(ctx, conn, node)
|
||||
close(done)
|
||||
}()
|
||||
select {
|
||||
case <-done:
|
||||
t.Fatal("the sweep judged the node's bucket while a send held it")
|
||||
case <-time.After(300 * time.Millisecond):
|
||||
}
|
||||
release()
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(20 * time.Second):
|
||||
t.Fatal("the sweep did not go on once the send let go")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,534 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/base64"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
sdkdecl "git.novox.be/novox/mesh-sdk/go/declaration"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/conditions"
|
||||
"github.com/novox/mesh-controller/internal/link"
|
||||
)
|
||||
|
||||
// contentOf is a deterministic content of n bytes that does not compress to nothing and holds no character JSON escapes.
|
||||
func contentOf(seed string, n int) string {
|
||||
var b strings.Builder
|
||||
for i := 0; b.Len() < n; i++ {
|
||||
fmt.Fprintf(&b, "%s-%07d-%x;", seed, i, i*2654435761%1000003)
|
||||
}
|
||||
return b.String()[:n]
|
||||
}
|
||||
|
||||
func aFileResource(id, content string) map[string]any {
|
||||
return map[string]any{"id": id, "type": "file", "path": "/etc/" + id, "mode": "0644", "content": content}
|
||||
}
|
||||
|
||||
// controlNodeShape is a declaration shaped like the control node's on 2026-10-21 (issue 1689), the sizes measured then: a bus users
|
||||
// file of 252,678 bytes, the vault's export of 138,475 bytes (compact), the tool runtime 21 KB, the packet filter 19 KB, the launcher
|
||||
// 14 KB, and about 600 small resources for the rest (about 286 KB). `more` scales the count of small ones: 1 is the shape, 1.4 forty
|
||||
// percent more of them.
|
||||
func controlNodeShape(more float64) []map[string]any {
|
||||
res := []map[string]any{
|
||||
aFileResource("nats.bus-users", contentOf("users", 252678)),
|
||||
aFileResource("mesh-vault.kept", contentOf("kept", 138475)),
|
||||
aFileResource("node-tools.runtime", contentOf("runtime", 21<<10)),
|
||||
aFileResource("nftables.filtering", contentOf("filter", 19<<10)),
|
||||
aFileResource("mesh-host.launcher", contentOf("launcher", 14<<10)),
|
||||
}
|
||||
for i := 0; i < int(600*more); i++ {
|
||||
res = append(res, aFileResource(fmt.Sprintf("module-%03d.config", i), contentOf(fmt.Sprintf("small%d", i), 380)))
|
||||
}
|
||||
return res
|
||||
}
|
||||
|
||||
func wireSizeOf(t *testing.T, s sendable) int {
|
||||
t.Helper()
|
||||
body, err := s.Body()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// The signature the controller's key makes is 64 bytes.
|
||||
msg, err := link.Envelope(body, make([]byte, 64))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return len(msg)
|
||||
}
|
||||
|
||||
// 1696/AC1: for a node that reports the capability, the two big files of a control-node-shaped declaration carry their digest and size
|
||||
// and no content, and the declaration is far under the bus's 1 MiB.
|
||||
func TestAFileOver64KiBTravelsByDigestToANodeThatReadsOne(t *testing.T) { // 1696/AC1
|
||||
users, kept := contentOf("users", 252678), contentOf("kept", 138475)
|
||||
s := sendable{fileReferences: true, Resources: []map[string]any{
|
||||
aFileResource("nats.bus-users", users), aFileResource("mesh-vault.kept", kept),
|
||||
aFileResource("small", "tiny\n"),
|
||||
}}
|
||||
body, err := s.Body()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var got struct {
|
||||
Resources []map[string]any `json:"resources"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &got); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for i, want := range []string{users, kept} {
|
||||
r := got.Resources[i]
|
||||
digest := sdkdecl.DigestOf([]byte(want))
|
||||
if _, has := r["content"]; has || r["digest"] != digest || r["size"] != float64(len(want)) {
|
||||
t.Errorf("%v goes with content %v, digest %v, size %v; want no content, %s and %d", r["id"], r["content"] != nil,
|
||||
r["digest"], r["size"], digest, len(want))
|
||||
}
|
||||
if r["path"] == nil || r["mode"] != "0644" {
|
||||
t.Errorf("%v lost its path or mode: %v", r["id"], r)
|
||||
}
|
||||
}
|
||||
if got.Resources[2]["content"] != "tiny\n" || got.Resources[2]["digest"] != nil {
|
||||
t.Errorf("a small file is not inline: %v", got.Resources[2])
|
||||
}
|
||||
if strings.Contains(string(body), users[:200]) || strings.Contains(string(body), kept[:200]) {
|
||||
t.Error("the content of a file that went by digest is in the body")
|
||||
}
|
||||
if n := wireSizeOf(t, s); n > 100<<10 {
|
||||
t.Errorf("the declaration is %d bytes on the wire, want under 100 KB", n)
|
||||
}
|
||||
// The objects to put are the contents, each under the hex of its digest, and a reference verifies against its object.
|
||||
objects, err := s.objects()
|
||||
if err != nil || len(objects) != 2 {
|
||||
t.Fatalf("objects: %v %v", objects, err)
|
||||
}
|
||||
for _, want := range []string{users, kept} {
|
||||
name, _ := sdkdecl.ObjectName(sdkdecl.DigestOf([]byte(want)))
|
||||
if string(objects[name]) != want {
|
||||
t.Errorf("the object %s does not hold the content", name[:12])
|
||||
}
|
||||
}
|
||||
named, err := namedObjects(body)
|
||||
if err != nil || len(named) != 2 {
|
||||
t.Fatalf("the objects the body names: %v %v", named, err)
|
||||
}
|
||||
for name := range objects {
|
||||
if _, ok := named[name]; !ok {
|
||||
t.Errorf("the body does not name the object %s it was sent with", name[:12])
|
||||
}
|
||||
if _, err := hex.DecodeString(name); err != nil || len(name) != 64 {
|
||||
t.Errorf("the object name %q is not 64 hex characters", name)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 1696/AC2: a node that does not report the capability is sent every content inline, as before, and has no object put.
|
||||
func TestANodeWithoutTheCapabilityIsSentEveryContentInline(t *testing.T) { // 1696/AC2
|
||||
users := contentOf("users", 252678)
|
||||
s := sendable{Resources: []map[string]any{aFileResource("nats.bus-users", users)}}
|
||||
body, err := s.Body()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var got struct {
|
||||
Resources []map[string]any `json:"resources"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &got); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got.Resources[0]["content"] != users || got.Resources[0]["digest"] != nil || got.Resources[0]["size"] != nil {
|
||||
t.Fatalf("a node that reads no references is not sent the content inline: %v", got.Resources[0]["digest"])
|
||||
}
|
||||
if objects, err := s.objects(); err != nil || len(objects) != 0 {
|
||||
t.Fatalf("objects for a node that reads none: %v %v", len(objects), err)
|
||||
}
|
||||
}
|
||||
|
||||
// 1696/AC3: only a file that may travel by digest does: one of exactly 64 KiB, a sealed one, one filled with secrets, one written
|
||||
// into another file, and a directory stay inline, whatever the node reads.
|
||||
func TestOnlyAFileThatMayTravelByDigestDoes(t *testing.T) { // 1696/AC3
|
||||
exact := contentOf("exact", 64<<10)
|
||||
big := contentOf("big", 64<<10+1)
|
||||
sealed := aFileResource("sealed", big)
|
||||
sealed["sealed"], sealed["content"] = "box", ""
|
||||
sealed["content"] = big // a sealed file's content is none; the body must pass it as it is, whatever it holds
|
||||
secret := aFileResource("secret", big)
|
||||
secret["secrets"] = map[string]any{"token": "x"}
|
||||
into := aFileResource("into", big)
|
||||
into["into"] = "/etc/other"
|
||||
s := sendable{fileReferences: true, Resources: []map[string]any{
|
||||
aFileResource("exact", exact), sealed, secret, into,
|
||||
{"id": "dir", "type": "directory", "path": "/etc/d", "content": big},
|
||||
aFileResource("big", big),
|
||||
}}
|
||||
body, err := s.Body()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var got struct {
|
||||
Resources []map[string]any `json:"resources"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &got); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for i, id := range []string{"exact", "sealed", "secret", "into", "dir"} {
|
||||
if got.Resources[i]["id"] != id || got.Resources[i]["digest"] != nil || got.Resources[i]["content"] == nil {
|
||||
t.Errorf("%s went by digest or lost its content: %v", id, got.Resources[i]["digest"])
|
||||
}
|
||||
}
|
||||
if big := got.Resources[5]; big["id"] != "big" || big["digest"] == nil || big["content"] != nil {
|
||||
t.Errorf("a plain file of 64 KiB and a byte did not go by digest: %v", big["digest"])
|
||||
}
|
||||
}
|
||||
|
||||
// A file given as bytes goes by digest on its decoded length, with the decoded bytes in the object.
|
||||
func TestAFileOfBytesTravelsByDigestOnItsDecodedLength(t *testing.T) {
|
||||
raw := []byte(contentOf("png", 70<<10))
|
||||
r := map[string]any{"id": "icon", "type": "file", "path": "/usr/share/icon.png", "bytes": stdBase64(raw)}
|
||||
s := sendable{fileReferences: true, Resources: []map[string]any{r}}
|
||||
objects, err := s.objects()
|
||||
if err != nil || len(objects) != 1 {
|
||||
t.Fatalf("%v %v", len(objects), err)
|
||||
}
|
||||
name, _ := sdkdecl.ObjectName(sdkdecl.DigestOf(raw))
|
||||
if string(objects[name]) != string(raw) {
|
||||
t.Fatal("the object is not the decoded bytes")
|
||||
}
|
||||
if r["bytes"] == nil {
|
||||
t.Fatal("the composed resource was changed in place")
|
||||
}
|
||||
}
|
||||
|
||||
type memObjects struct {
|
||||
have map[string][]byte
|
||||
marked map[string]bool
|
||||
puts []string
|
||||
fail map[string]error
|
||||
}
|
||||
|
||||
func (m *memObjects) Has(_ context.Context, name string) (bool, bool, error) {
|
||||
_, ok := m.have[name]
|
||||
return ok, m.marked[name], nil
|
||||
}
|
||||
func (m *memObjects) Put(_ context.Context, name string, c []byte) error {
|
||||
if err := m.fail[name]; err != nil {
|
||||
return err
|
||||
}
|
||||
m.puts = append(m.puts, name)
|
||||
m.have[name] = c
|
||||
delete(m.marked, name)
|
||||
return nil
|
||||
}
|
||||
|
||||
// 1696/AC4: an object already in the bucket is not put again, one absent is, and a put that fails stops the send, naming the object.
|
||||
func TestObjectsAreInTheBucketBeforeTheDeclarationIsSent(t *testing.T) { // 1696/AC4
|
||||
objects := map[string][]byte{strings.Repeat("a", 64): []byte("one"), strings.Repeat("b", 64): []byte("two")}
|
||||
b := &memObjects{have: map[string][]byte{strings.Repeat("a", 64): []byte("one")}}
|
||||
if err := putObjects(t.Context(), b, objects); err != nil || len(b.puts) != 1 || b.puts[0] != strings.Repeat("b", 64) {
|
||||
t.Fatalf("put %v (%v), want the one absent", b.puts, err)
|
||||
}
|
||||
// An object the sweep marked unnamed is put again, which erases the mark: a send relies on it.
|
||||
b = &memObjects{have: map[string][]byte{strings.Repeat("a", 64): []byte("one"), strings.Repeat("b", 64): []byte("two")},
|
||||
marked: map[string]bool{strings.Repeat("a", 64): true}}
|
||||
if err := putObjects(t.Context(), b, objects); err != nil || len(b.puts) != 1 || b.puts[0] != strings.Repeat("a", 64) || b.marked[strings.Repeat("a", 64)] {
|
||||
t.Fatalf("a marked object: put %v (%v) marked %v, want it put again and the mark gone", b.puts, err, b.marked)
|
||||
}
|
||||
b = &memObjects{have: map[string][]byte{}, fail: map[string]error{strings.Repeat("a", 64): errors.New("permissions violation")}}
|
||||
err := putObjects(t.Context(), b, objects)
|
||||
if err == nil || !strings.Contains(err.Error(), "aaaaaaaaaaaa") || !strings.Contains(err.Error(), "permissions violation") {
|
||||
t.Fatalf("a failed put: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// fallbackDelivery refuses the first declaration it is sent with the bucket's failure, as a bus that has not yet been given the
|
||||
// controller's grant does, and takes the second.
|
||||
type fallbackDelivery struct {
|
||||
bodies [][]byte
|
||||
first error
|
||||
}
|
||||
|
||||
func (f *fallbackDelivery) grant(context.Context, []readyNode) error { return nil }
|
||||
func (f *fallbackDelivery) declare(_ context.Context, _ readyNode, body []byte) (string, error) {
|
||||
f.bodies = append(f.bodies, body)
|
||||
if len(f.bodies) == 1 && f.first != nil {
|
||||
return "", fmt.Errorf("x: %w: %w", errFilesNotPut, f.first)
|
||||
}
|
||||
return "digest", nil
|
||||
}
|
||||
|
||||
// 1696/AC5: when the bucket cannot be written yet (the grant travels in the declaration of the machine holding the bus), that send goes
|
||||
// with every content inline and is made; a send that fails for another reason is not retried inline.
|
||||
func TestASendWhoseFilesCannotBePutGoesInlineOnce(t *testing.T) { // 1696/AC5
|
||||
users := contentOf("users", 100<<10)
|
||||
s := readyNode{node: "novox", declared: sendable{fileReferences: true, Resources: []map[string]any{aFileResource("nats.bus-users", users)}}}
|
||||
f := &fallbackDelivery{first: errors.New("timeout")}
|
||||
digests, err := deliver(t.Context(), f, "novox", []readyNode{s})
|
||||
if err != nil || digests["novox"] != "digest" || len(f.bodies) != 2 {
|
||||
t.Fatalf("%v %v %d", digests, err, len(f.bodies))
|
||||
}
|
||||
if strings.Contains(string(f.bodies[0]), users[:100]) || !strings.Contains(string(f.bodies[1]), users[:100]) {
|
||||
t.Error("the first body should name the file by digest and the second carry it inline")
|
||||
}
|
||||
other := &fallbackDelivery{}
|
||||
other.first = nil
|
||||
failing := &failingDelivery{err: errors.New("the bus is down")}
|
||||
if _, err := deliver(t.Context(), failing, "novox", []readyNode{s}); err == nil || failing.calls != 1 {
|
||||
t.Fatalf("a failure of another kind: %v after %d send(s)", err, failing.calls)
|
||||
}
|
||||
}
|
||||
|
||||
type failingDelivery struct {
|
||||
err error
|
||||
calls int
|
||||
}
|
||||
|
||||
func (f *failingDelivery) grant(context.Context, []readyNode) error { return nil }
|
||||
func (f *failingDelivery) declare(context.Context, readyNode, []byte) (string, error) {
|
||||
f.calls++
|
||||
return "", f.err
|
||||
}
|
||||
|
||||
// A bucket for the sweep.
|
||||
type sweepBucketFake struct {
|
||||
objects map[string]time.Time // zero: named at the last look
|
||||
deleted []string
|
||||
}
|
||||
|
||||
func (b *sweepBucketFake) Objects(context.Context) ([]bucketObject, error) {
|
||||
var out []bucketObject
|
||||
for n, at := range b.objects {
|
||||
out = append(out, bucketObject{Name: n, UnnamedSince: at})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
func (b *sweepBucketFake) MarkUnnamed(_ context.Context, n string, at time.Time) error {
|
||||
b.objects[n] = at
|
||||
return nil
|
||||
}
|
||||
func (b *sweepBucketFake) ClearUnnamed(_ context.Context, n string) error {
|
||||
b.objects[n] = time.Time{}
|
||||
return nil
|
||||
}
|
||||
func (b *sweepBucketFake) Delete(_ context.Context, n string, markedAt time.Time) (bool, error) {
|
||||
if at, ok := b.objects[n]; !ok || !at.Equal(markedAt) {
|
||||
return false, nil // put again since it was judged: its mark is gone
|
||||
}
|
||||
delete(b.objects, n)
|
||||
b.deleted = append(b.deleted, n)
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// 1696/AC6: an object a latest declaration still names is not removed, however long; one unnamed is marked at the first look and removed
|
||||
// only when the mark is 24 hours old; one named again loses its mark.
|
||||
func TestAnObjectIsRemovedADayAfterNoLatestDeclarationNamesIt(t *testing.T) { // 1696/AC6
|
||||
named, unnamed, back := strings.Repeat("1", 64), strings.Repeat("2", 64), strings.Repeat("3", 64)
|
||||
t0 := time.Date(2026, 10, 21, 12, 0, 0, 0, time.UTC)
|
||||
b := &sweepBucketFake{objects: map[string]time.Time{named: t0.Add(-100 * time.Hour), unnamed: {}, back: t0.Add(-5 * time.Hour)}}
|
||||
named3 := map[string]bool{named: true, back: true}
|
||||
r, err := sweepDeclarationFiles(t.Context(), b, named3, t0)
|
||||
if err != nil || r.Marked != 1 || r.Cleared != 2 || r.Removed != 0 || len(b.deleted) != 0 {
|
||||
t.Fatalf("first look: %+v %v %v", r, err, b.deleted)
|
||||
}
|
||||
if !b.objects[named].IsZero() || !b.objects[back].IsZero() || !b.objects[unnamed].Equal(t0) {
|
||||
t.Fatalf("marks after the first look: %v", b.objects)
|
||||
}
|
||||
// Still unnamed 23 hours 59 minutes later: kept. A day after the mark: removed. The named ones stay for ever.
|
||||
if r, err = sweepDeclarationFiles(t.Context(), b, map[string]bool{named: true, back: true}, t0.Add(24*time.Hour-time.Minute)); err != nil ||
|
||||
r.Removed != 0 || len(b.deleted) != 0 {
|
||||
t.Fatalf("a minute short of a day: %+v %v %v", r, err, b.deleted)
|
||||
}
|
||||
if r, err = sweepDeclarationFiles(t.Context(), b, map[string]bool{named: true, back: true}, t0.Add(24*time.Hour)); err != nil ||
|
||||
r.Removed != 1 || len(b.deleted) != 1 || b.deleted[0] != unnamed {
|
||||
t.Fatalf("a day after: %+v %v %v", r, err, b.deleted)
|
||||
}
|
||||
if _, ok := b.objects[named]; !ok {
|
||||
t.Fatal("an object a latest declaration names was removed")
|
||||
}
|
||||
// An object named again before the day is up loses its mark and is not removed.
|
||||
b = &sweepBucketFake{objects: map[string]time.Time{unnamed: t0}}
|
||||
if _, err := sweepDeclarationFiles(t.Context(), b, map[string]bool{unnamed: true}, t0.Add(23*time.Hour)); err != nil || !b.objects[unnamed].IsZero() {
|
||||
t.Fatalf("named again: %v %v", b.objects, err)
|
||||
}
|
||||
}
|
||||
|
||||
// What a node's latest declaration names is read from the signed envelope as stored on the bus; an unreadable one is an error and
|
||||
// never "names none", which would mark an object a node needs.
|
||||
func TestTheObjectsALatestDeclarationNamesAreReadFromTheBus(t *testing.T) {
|
||||
s := sendable{fileReferences: true, Resources: []map[string]any{aFileResource("a", contentOf("a", 70<<10))}}
|
||||
body, _ := s.Body()
|
||||
msg, _ := link.Envelope(body, []byte("sig"))
|
||||
named, err := namedByLatest("novox", msg)
|
||||
name, _ := sdkdecl.ObjectName(sdkdecl.DigestOf([]byte(contentOf("a", 70<<10))))
|
||||
if err != nil || len(named) != 1 || !named[name] {
|
||||
t.Fatalf("%v %v", named, err)
|
||||
}
|
||||
if none, err := namedByLatest("ace", []byte(`{"declaration":"e30=","signature":""}`)); err != nil || len(none) != 0 {
|
||||
t.Fatalf("a declaration naming none: %v %v", none, err)
|
||||
}
|
||||
if _, err := namedByLatest("novox", []byte("not json")); err == nil {
|
||||
t.Fatal("an unreadable latest declaration was read as naming none")
|
||||
}
|
||||
}
|
||||
|
||||
// 1698/AC2 (ADR 0486 §9): a declaration shaped like the control node's, for a node that reads references, stays under half of the
|
||||
// bus's largest message on the wire; the inline form of the same does not fit at all; and the same shape with 40 percent more resources
|
||||
// fails the test's own limit, so a change that eats the room fails in a pull request and not on the live mesh.
|
||||
func TestTheControlNodesDeclarationInTheReferenceFormStaysUnderHalfTheBusLimit(t *testing.T) { // 1698/AC2
|
||||
const limit = 1048576
|
||||
shape := sendable{fileReferences: true, Resources: controlNodeShape(1)}
|
||||
n := wireSizeOf(t, shape)
|
||||
t.Logf("the control node's shape in the reference form: %d bytes on the wire, %.1f%% of %d", n, 100*float64(n)/limit, limit)
|
||||
if n > limit/2 {
|
||||
t.Errorf("the reference form is %d bytes on the wire, over half (%d) of the bus's largest message", n, limit/2)
|
||||
}
|
||||
if inline := wireSizeOf(t, sendable{Resources: controlNodeShape(1)}); inline < limit*3/4 {
|
||||
t.Errorf("the inline form is %d bytes, which is not the shape that was over the limit", inline)
|
||||
}
|
||||
grown := wireSizeOf(t, sendable{fileReferences: true, Resources: controlNodeShape(1.4)})
|
||||
if grown <= limit/2 {
|
||||
t.Errorf("the shape with 40%% more resources is %d bytes, which the test would let pass", grown)
|
||||
}
|
||||
}
|
||||
|
||||
func stdBase64(b []byte) string { return base64.StdEncoding.EncodeToString(b) }
|
||||
|
||||
// A sweep that read an object's mark before a send put it again must not remove it: the object, now without the mark, is left.
|
||||
func TestASweepDoesNotRemoveAnObjectASendPutAgainAfterItWasJudged(t *testing.T) { // 1696/AC8
|
||||
n := strings.Repeat("4", 64)
|
||||
t0 := time.Date(2026, 10, 21, 12, 0, 0, 0, time.UTC)
|
||||
b := &sweepBucketFake{objects: map[string]time.Time{n: t0}}
|
||||
judged := bucketObject{Name: n, UnnamedSince: t0}
|
||||
b.objects[n] = time.Time{} // the send put it again: no mark
|
||||
if gone, err := b.Delete(t.Context(), judged.Name, judged.UnnamedSince); err != nil || gone || len(b.deleted) != 0 {
|
||||
t.Fatalf("an object put again was removed: %v %v %v", gone, err, b.deleted)
|
||||
}
|
||||
}
|
||||
|
||||
// One node's bucket is held by one act at a time in this process: the sweep waits for a send that has put its objects and not yet
|
||||
// published the declaration that names them (1696/AC9).
|
||||
func TestTheSweepOfANodesBucketWaitsForASendThatHasPutItsObjects(t *testing.T) { // 1696/AC9
|
||||
unlock := lockFilesOf("lock-test-node")
|
||||
got := make(chan struct{})
|
||||
go func() {
|
||||
defer lockFilesOf("lock-test-node")()
|
||||
close(got)
|
||||
}()
|
||||
select {
|
||||
case <-got:
|
||||
t.Fatal("the sweep took the node's bucket while a send held it")
|
||||
case <-time.After(50 * time.Millisecond):
|
||||
}
|
||||
other := lockFilesOf("lock-test-other-node") // another node's bucket is not held
|
||||
other()
|
||||
unlock()
|
||||
select {
|
||||
case <-got:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("the sweep did not take the node's bucket once the send was done")
|
||||
}
|
||||
}
|
||||
|
||||
// A put that fails send after send raises a condition; the first failure does not (the grant it waits for travels in that very
|
||||
// declaration), and a put that works clears it (1696/AC10).
|
||||
func TestAPutThatFailsSendAfterSendRaisesAConditionAndClearsWhenItWorks(t *testing.T) { // 1696/AC10
|
||||
store := conditions.NewInMemory()
|
||||
k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store,
|
||||
Teller: &conditions.Told{}, Now: func() time.Time { return time.Date(2026, 10, 21, 14, 0, 0, 0, time.UTC) }})
|
||||
t.Cleanup(func() { k.Close(t.Context()) })
|
||||
was := conditionsFrom
|
||||
conditionsFrom = k
|
||||
t.Cleanup(func() { conditionsFrom = was })
|
||||
key := filesNotPutObservation("cond-node", "").Key()
|
||||
open := func() bool {
|
||||
_, found, err := conditions.ReadOne(t.Context(), store, key)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return found
|
||||
}
|
||||
boom := errors.New("permissions violation")
|
||||
notePut(t.Context(), "cond-node", boom)
|
||||
if open() {
|
||||
t.Fatal("the first failure raised the condition")
|
||||
}
|
||||
notePut(t.Context(), "cond-node", boom)
|
||||
if !open() {
|
||||
t.Fatal("the second failure in a row did not raise the condition")
|
||||
}
|
||||
notePut(t.Context(), "cond-node", nil)
|
||||
if open() {
|
||||
t.Fatal("a put that worked did not clear the condition")
|
||||
}
|
||||
notePut(t.Context(), "cond-node", boom)
|
||||
if open() {
|
||||
t.Fatal("the count did not start again after a put that worked")
|
||||
}
|
||||
}
|
||||
|
||||
// aBusThatSays is a bus that keeps the declarations published to it and, at each, asks whether the node's bucket is held.
|
||||
type aBusThatSays struct {
|
||||
link.Bus
|
||||
published chan string
|
||||
heldAtPublish func() bool
|
||||
}
|
||||
|
||||
func (b *aBusThatSays) PublishDeclaration(_ context.Context, node string, _ []byte) error {
|
||||
if b.heldAtPublish != nil && !b.heldAtPublish() {
|
||||
return errors.New("published while the node's bucket was not held")
|
||||
}
|
||||
b.published <- node
|
||||
return errors.New("stop here: nothing is recorded")
|
||||
}
|
||||
|
||||
type aSignerOfZeros struct{}
|
||||
|
||||
func (aSignerOfZeros) Sign(context.Context, []byte) ([]byte, error) { return make([]byte, 64), nil }
|
||||
|
||||
// whileHeld says whether the node's lock is held by someone right now, without keeping it.
|
||||
func whileHeld(node string) bool {
|
||||
got := make(chan struct{})
|
||||
go func() {
|
||||
defer lockFilesOf(node)()
|
||||
close(got)
|
||||
}()
|
||||
select {
|
||||
case <-got:
|
||||
return false
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
// held: the goroutine stays waiting and takes and releases the lock once it is let go.
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
// 1696/AC9: a send holds the node's lock from the put to the publication, so the sweep never removes an object between them. The
|
||||
// test goes through overTheBus.declare: while another holds the lock the declaration is not published, and at the publication the
|
||||
// lock is held.
|
||||
func TestASendHoldsTheNodesLockFromThePutToThePublication(t *testing.T) { // 1696/AC9
|
||||
const node = "lock-send-node"
|
||||
bus := &aBusThatSays{published: make(chan string, 1)}
|
||||
bus.heldAtPublish = func() bool { return whileHeld(node) }
|
||||
send := overTheBus{server: link.SendingOn(bus), signer: aSignerOfZeros{}}
|
||||
release := lockFilesOf(node) // a sweep is judging this node's bucket
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
_, err := send.declare(t.Context(), readyNode{node: node}, []byte(`{}`))
|
||||
done <- err
|
||||
}()
|
||||
select {
|
||||
case n := <-bus.published:
|
||||
t.Fatalf("%s was sent its declaration while the sweep held the node's bucket", n)
|
||||
case <-time.After(150 * time.Millisecond):
|
||||
}
|
||||
release()
|
||||
select {
|
||||
case <-bus.published:
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("the send did not go on once the sweep let go")
|
||||
}
|
||||
if err := <-done; err == nil || !strings.Contains(err.Error(), "nothing is recorded") {
|
||||
t.Fatalf("the send did not reach the publication (a lock held at it): %v", err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,96 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/conditions"
|
||||
"github.com/novox/mesh-controller/internal/link"
|
||||
)
|
||||
|
||||
// **A declaration is one bus message, and the bus's largest message is a cliff** (novox/hq issue 1689). On 2026-10-21
|
||||
// the control node's declaration passed it by 2 KB and nothing reached that node until two assignments were undone.
|
||||
// So the size is measured at every send: a warning condition from 75% of the limit (ADR 0486 §8, issue 1698: it was 90% while every file travelled inline), urgent over it, cleared when
|
||||
// the declaration is back under 75%. The limit is the connection's own, not a constant.
|
||||
|
||||
// kindDeclarationNearLimit is a declaration at or near the bus's largest message.
|
||||
const kindDeclarationNearLimit = "declaration-near-limit"
|
||||
|
||||
// declarationRoomSource is the condition's source: the sends.
|
||||
const declarationRoomSource = "sends"
|
||||
|
||||
func declarationRoomObservation(node string, size, limit int64) conditions.Observation {
|
||||
o := conditions.Observation{Scope: conditions.ScopeMachine, ID: node, Kind: kindDeclarationNearLimit,
|
||||
Token: kindDeclarationNearLimit, Machine: node, Severity: conditions.Warning, Source: declarationRoomSource}
|
||||
if size > limit {
|
||||
o.Severity = conditions.Urgent
|
||||
}
|
||||
o.Summary = fmt.Sprintf("%s's declaration is %d bytes on the wire, %d%% of the bus's largest message of %d bytes",
|
||||
node, size, size*100/limit, limit)
|
||||
if size > limit {
|
||||
o.Summary += ": over it, so it cannot be sent"
|
||||
}
|
||||
return o
|
||||
}
|
||||
|
||||
func init() {
|
||||
link.DeclarationMeasured = func(ctx context.Context, node string, size, limit int64) {
|
||||
settleDeclarationRoom(ctx, conditionsFrom, node, size, limit)
|
||||
}
|
||||
}
|
||||
|
||||
// settleDeclarationRoom raises or clears the condition of a declaration's size. A nil keeper (a tool, not the
|
||||
// serving controller) says nothing.
|
||||
func settleDeclarationRoom(ctx context.Context, keeper *conditions.Keeper, node string, size, limit int64) {
|
||||
if keeper == nil || limit <= 0 {
|
||||
return
|
||||
}
|
||||
obs := declarationRoomObservation(node, size, limit)
|
||||
if float64(size) < link.NearTheLimit*float64(limit) {
|
||||
if _, err := keeper.Clear(ctx, obs.Key(), fmt.Sprintf("%s's declaration is %d bytes, under the limit's margin", node, size)); err != nil {
|
||||
fmt.Printf("the condition %s could not be cleared: %v\n", obs.Key(), err)
|
||||
}
|
||||
return
|
||||
}
|
||||
fmt.Printf("NEAR THE LIMIT: %s\n", obs.Summary)
|
||||
if _, err := keeper.Observe(ctx, obs); err != nil {
|
||||
fmt.Printf("the condition %s could not be raised: %v\n", obs.Key(), err)
|
||||
}
|
||||
}
|
||||
|
||||
// compactKept writes the vault's export compact. The composition (the SDK's keptFile) writes it indented, and its 334
|
||||
// sealed secrets were 164,763 bytes of it against 138,474 compact: a fifth of the declaration's room, in whitespace,
|
||||
// base64-encoded again on the wire. Every reader parses JSON, so the spacing is nobody's. A resource that is not
|
||||
// the export, or whose content is not JSON, is left as it is. The resources are copied, never changed in place.
|
||||
func compactKept(resources []map[string]any) []map[string]any {
|
||||
out := make([]map[string]any, len(resources))
|
||||
for i, r := range resources {
|
||||
out[i] = r
|
||||
id, _ := r["id"].(string)
|
||||
path, _ := r["path"].(string)
|
||||
content, ok := r["content"].(string)
|
||||
if !ok || (id != "kept" && !strings.HasSuffix(id, ".kept")) || !strings.HasSuffix(path, "/export.json") {
|
||||
continue
|
||||
}
|
||||
var dst bytes.Buffer
|
||||
if json.Compact(&dst, []byte(content)) != nil {
|
||||
continue
|
||||
}
|
||||
if strings.HasSuffix(content, "\n") {
|
||||
dst.WriteByte('\n')
|
||||
}
|
||||
if dst.String() == content {
|
||||
continue
|
||||
}
|
||||
c := make(map[string]any, len(r))
|
||||
for k, v := range r {
|
||||
c[k] = v
|
||||
}
|
||||
c["content"] = dst.String()
|
||||
out[i] = c
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,71 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/conditions"
|
||||
)
|
||||
|
||||
// The vault's export is sent compact (issue 1689): same document, no indentation, one rewrite.
|
||||
func TestTheVaultsExportIsSentCompactAndNothingElseIsTouched(t *testing.T) {
|
||||
indented := "{\n \"export\": 1,\n \"kept\": [\n {\n \"name\": \"a\"\n }\n ]\n}\n"
|
||||
other := "{\n \"x\": 1\n}\n"
|
||||
in := []map[string]any{
|
||||
{"id": "mesh-vault.kept", "type": "file", "path": "/srv/vault/export.json", "mode": "0600", "content": indented},
|
||||
{"id": "mesh-vault.config", "type": "file", "path": "/srv/vault/config.json", "content": other},
|
||||
{"id": "other.kept", "type": "file", "path": "/srv/other/notes.json", "content": other},
|
||||
}
|
||||
out := compactKept(in)
|
||||
want := "{\"export\":1,\"kept\":[{\"name\":\"a\"}]}\n"
|
||||
if out[0]["content"] != want || out[0]["mode"] != "0600" || out[0]["path"] != "/srv/vault/export.json" {
|
||||
t.Fatalf("the export is %q (%v), expected %q with its mode and path kept", out[0]["content"], out[0], want)
|
||||
}
|
||||
if out[1]["content"] != other || out[2]["content"] != other {
|
||||
t.Fatalf("a file that is not the export was rewritten: %v %v", out[1], out[2])
|
||||
}
|
||||
if in[0]["content"] != indented {
|
||||
t.Fatalf("the composed resource was changed in place")
|
||||
}
|
||||
if again := compactKept(out); again[0]["content"] != want {
|
||||
t.Fatalf("compacting twice changed it: %q", again[0]["content"])
|
||||
}
|
||||
}
|
||||
|
||||
// From 75% of the bus's largest message (ADR 0486 §8, issue 1698) the declaration is a warning; over it, urgent; back under 75%, cleared.
|
||||
func TestADeclarationNearTheBusLimitRaisesAConditionAndClearsWhenSmaller(t *testing.T) {
|
||||
store := conditions.NewInMemory()
|
||||
k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store,
|
||||
Teller: &conditions.Told{}, Now: func() time.Time { return time.Date(2026, 10, 21, 14, 0, 0, 0, time.UTC) }})
|
||||
t.Cleanup(func() { k.Close(t.Context()) })
|
||||
const limit = 1000
|
||||
key := declarationRoomObservation("novox", 0, limit).Key()
|
||||
read := func() (conditions.Condition, bool) {
|
||||
t.Helper()
|
||||
c, found, err := conditions.ReadOne(t.Context(), store, key)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return c, found
|
||||
}
|
||||
|
||||
settleDeclarationRoom(t.Context(), k, "novox", 749, limit)
|
||||
if _, found := read(); found {
|
||||
t.Fatal("a declaration at 74.9% raised a condition")
|
||||
}
|
||||
settleDeclarationRoom(t.Context(), k, "novox", 750, limit)
|
||||
c, found := read()
|
||||
if !found || c.Severity != conditions.Warning || !strings.Contains(c.Summary, "750 bytes") ||
|
||||
!strings.Contains(c.Summary, "1000") || c.Subject.Machine != "novox" {
|
||||
t.Fatalf("at 75%% the condition is %v %+v", found, c)
|
||||
}
|
||||
settleDeclarationRoom(t.Context(), k, "novox", 1001, limit)
|
||||
if c, _ := read(); c.Severity != conditions.Urgent {
|
||||
t.Fatalf("over the limit the condition is %q, expected urgent", c.Severity)
|
||||
}
|
||||
settleDeclarationRoom(t.Context(), k, "novox", 700, limit)
|
||||
if _, found := read(); found {
|
||||
t.Fatal("the condition stayed after the declaration fell under 75%")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,273 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
"github.com/novox/mesh-controller/internal/link"
|
||||
)
|
||||
|
||||
// **An object is removed a day after no node's latest declaration names it** (novox/hq ADR 0486 §3, issue 1696), so a node that is
|
||||
// asleep and receives its latest declaration later still finds every object that declaration names. What each node's latest
|
||||
// declaration names is read from the bus itself, the last message of that node's subject of the stream NODES: nothing else
|
||||
// is kept of it. When an object is first found named by none, the time is written in its own metadata (declarationUnnamed);
|
||||
// found named again, that is erased; found unnamed for a day, it is removed. Each node has a bucket of its own, judged against that
|
||||
// node's latest declaration alone. Objects are immutable and regenerable from the
|
||||
// controller's state, so a removal loses nothing.
|
||||
|
||||
// declarationUnnamed is the key of an object's metadata holding when it was first found named by no node's latest declaration.
|
||||
const declarationUnnamed = "unnamed-since"
|
||||
|
||||
// sweepEvery is how often the bucket is judged: the removal is exact to within this.
|
||||
const sweepEvery = time.Hour
|
||||
|
||||
// bucketObject is one object of the bucket as the sweep sees it.
|
||||
type bucketObject struct {
|
||||
Name string
|
||||
// UnnamedSince is when it was first found named by none, zero when it was named at the last look.
|
||||
UnnamedSince time.Time
|
||||
}
|
||||
|
||||
// sweptBucket is what the sweep needs of the bucket.
|
||||
type sweptBucket interface {
|
||||
Objects(ctx context.Context) ([]bucketObject, error)
|
||||
MarkUnnamed(ctx context.Context, name string, since time.Time) error
|
||||
ClearUnnamed(ctx context.Context, name string) error
|
||||
// Delete removes the object if it still carries the mark it was judged by, and says whether it did: a send that put it again
|
||||
// since has erased the mark, and the object is left.
|
||||
Delete(ctx context.Context, name string, markedAt time.Time) (bool, error)
|
||||
}
|
||||
|
||||
// filesSweep says what a sweep did.
|
||||
type filesSweep struct{ Kept, Marked, Cleared, Removed int }
|
||||
|
||||
// sweepDeclarationFiles judges every object against the names of the latest declarations: named ones are kept (and a
|
||||
// mark of being unnamed is erased), unnamed ones are marked at the first look and removed once the mark is a day old.
|
||||
func sweepDeclarationFiles(ctx context.Context, bucket sweptBucket, named map[string]bool, now time.Time) (filesSweep, error) {
|
||||
var r filesSweep
|
||||
objects, err := bucket.Objects(ctx)
|
||||
if err != nil {
|
||||
return r, err
|
||||
}
|
||||
sort.Slice(objects, func(i, j int) bool { return objects[i].Name < objects[j].Name })
|
||||
for _, o := range objects {
|
||||
switch {
|
||||
case named[o.Name]:
|
||||
r.Kept++
|
||||
if !o.UnnamedSince.IsZero() {
|
||||
if err := bucket.ClearUnnamed(ctx, o.Name); err != nil {
|
||||
return r, fmt.Errorf("the object %s: %w", o.Name[:12], err)
|
||||
}
|
||||
r.Cleared++
|
||||
}
|
||||
case o.UnnamedSince.IsZero():
|
||||
if err := bucket.MarkUnnamed(ctx, o.Name, now); err != nil {
|
||||
return r, fmt.Errorf("the object %s: %w", o.Name[:12], err)
|
||||
}
|
||||
r.Marked++
|
||||
case now.Sub(o.UnnamedSince) >= broker.DeclarationFilesKeptFor:
|
||||
gone, err := bucket.Delete(ctx, o.Name, o.UnnamedSince)
|
||||
if err != nil {
|
||||
return r, fmt.Errorf("the object %s: %w", o.Name[:12], err)
|
||||
}
|
||||
if gone {
|
||||
r.Removed++
|
||||
} else {
|
||||
r.Kept++
|
||||
}
|
||||
default:
|
||||
r.Kept++
|
||||
}
|
||||
}
|
||||
return r, nil
|
||||
}
|
||||
|
||||
// namedByLatest is the name of every object this latest declaration of a node names. A body that cannot be read is an error, never
|
||||
// "names none": an object the node still needs would be marked for removal.
|
||||
func namedByLatest(node string, signed []byte) (map[string]bool, error) {
|
||||
var s link.Signed
|
||||
if err := json.Unmarshal(signed, &s); err != nil {
|
||||
return nil, fmt.Errorf("%s's latest declaration is not readable: %w", node, err)
|
||||
}
|
||||
names, err := namedObjects(s.Declaration)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%s's latest declaration is not readable: %w", node, err)
|
||||
}
|
||||
named := make(map[string]bool, len(names))
|
||||
for name := range names {
|
||||
named[name] = true
|
||||
}
|
||||
return named, nil
|
||||
}
|
||||
|
||||
// latestDeclarations are the last signed declarations of the nodes, by node, as the stream NODES holds them. A node with none has
|
||||
// no entry.
|
||||
func latestDeclarations(ctx context.Context, conn *nats.Conn, nodes []string) (map[string][]byte, error) {
|
||||
js, err := conn.JetStream(nats.Context(ctx))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := map[string][]byte{}
|
||||
for _, node := range nodes {
|
||||
msg, err := js.GetLastMsg("NODES", link.DeclareSubject(node), nats.Context(ctx))
|
||||
if errors.Is(err, nats.ErrMsgNotFound) {
|
||||
continue
|
||||
}
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the latest declaration of %s: %w", node, err)
|
||||
}
|
||||
out[node] = msg.Data
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// busSweptBucket is the bucket on the controller's connection.
|
||||
type busSweptBucket struct{ store jetstream.ObjectStore }
|
||||
|
||||
func (b busSweptBucket) Objects(ctx context.Context) ([]bucketObject, error) {
|
||||
infos, err := b.store.List(ctx)
|
||||
if errors.Is(err, jetstream.ErrNoObjectsFound) {
|
||||
return nil, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]bucketObject, 0, len(infos))
|
||||
for _, info := range infos {
|
||||
o := bucketObject{Name: info.Name}
|
||||
if at, err := time.Parse(time.RFC3339, info.Metadata[declarationUnnamed]); err == nil {
|
||||
o.UnnamedSince = at
|
||||
}
|
||||
out = append(out, o)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (b busSweptBucket) meta(ctx context.Context, name string, change func(map[string]string)) error {
|
||||
info, err := b.store.GetInfo(ctx, name)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
meta := info.ObjectMeta
|
||||
kept := map[string]string{}
|
||||
for k, v := range meta.Metadata {
|
||||
kept[k] = v
|
||||
}
|
||||
change(kept)
|
||||
meta.Metadata = kept
|
||||
return b.store.UpdateMeta(ctx, name, meta)
|
||||
}
|
||||
|
||||
func (b busSweptBucket) MarkUnnamed(ctx context.Context, name string, since time.Time) error {
|
||||
return b.meta(ctx, name, func(m map[string]string) { m[declarationUnnamed] = since.UTC().Format(time.RFC3339) })
|
||||
}
|
||||
|
||||
func (b busSweptBucket) ClearUnnamed(ctx context.Context, name string) error {
|
||||
return b.meta(ctx, name, func(m map[string]string) { delete(m, declarationUnnamed) })
|
||||
}
|
||||
|
||||
func (b busSweptBucket) Delete(ctx context.Context, name string, markedAt time.Time) (bool, error) {
|
||||
info, err := b.store.GetInfo(ctx, name)
|
||||
if errors.Is(err, jetstream.ErrObjectNotFound) {
|
||||
return false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
if at, perr := time.Parse(time.RFC3339, info.Metadata[declarationUnnamed]); perr != nil || !at.Equal(markedAt) {
|
||||
return false, nil
|
||||
}
|
||||
return true, b.store.Delete(ctx, name)
|
||||
}
|
||||
|
||||
// sweepingDeclarationFiles judges the bucket every sweepEvery for as long as this controller serves and holds the lease.
|
||||
func sweepingDeclarationFiles(ctx context.Context, inv *inventory.Inventory, conn *nats.Conn) {
|
||||
tick := time.NewTicker(sweepEvery)
|
||||
defer tick.Stop()
|
||||
failing := ""
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-tick.C:
|
||||
}
|
||||
if epoch, err := theLease.epoch(ctx); err != nil || epoch == 0 {
|
||||
continue
|
||||
}
|
||||
why := sweepOnce(ctx, inv, conn)
|
||||
if why != failing {
|
||||
if why != "" {
|
||||
fmt.Printf("the bucket of declaration files cannot be swept: %s\n", why)
|
||||
} else {
|
||||
fmt.Printf("the bucket of declaration files is swept again\n")
|
||||
}
|
||||
failing = why
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// sweepOnce judges the bucket of every node against that node's latest declaration, one bucket at a time under the node's lock, with
|
||||
// the declaration read again under it: a send that put an object and published its declaration is seen, and one that has not yet
|
||||
// is waiting for the lock.
|
||||
func sweepOnce(ctx context.Context, inv *inventory.Inventory, conn *nats.Conn) string {
|
||||
nodes, err := inv.Nodes(ctx)
|
||||
if err != nil {
|
||||
return fmt.Sprintf("the nodes cannot be read: %v", err)
|
||||
}
|
||||
var problems []string
|
||||
var total filesSweep
|
||||
for _, n := range nodes {
|
||||
r, err := sweepNode(ctx, conn, n.Name)
|
||||
total.Kept, total.Marked, total.Cleared, total.Removed = total.Kept+r.Kept, total.Marked+r.Marked, total.Cleared+r.Cleared, total.Removed+r.Removed
|
||||
if err != nil {
|
||||
problems = append(problems, fmt.Sprintf("%s: %v", n.Name, err))
|
||||
}
|
||||
}
|
||||
if total.Removed+total.Marked+total.Cleared > 0 {
|
||||
fmt.Printf("the buckets of declaration files: %d object(s) kept, %d found unnamed, %d named again, %d removed\n",
|
||||
total.Kept, total.Marked, total.Cleared, total.Removed)
|
||||
}
|
||||
return strings.Join(problems, "; ")
|
||||
}
|
||||
|
||||
func sweepNode(ctx context.Context, conn *nats.Conn, node string) (filesSweep, error) {
|
||||
working, cancel := context.WithTimeout(ctx, 2*time.Minute)
|
||||
defer cancel()
|
||||
js, err := jetstream.New(conn)
|
||||
if err != nil {
|
||||
return filesSweep{}, err
|
||||
}
|
||||
bucket, err := broker.DeclarationFilesBucketOf(node)
|
||||
if err != nil {
|
||||
return filesSweep{}, fmt.Errorf("the bucket of node %q: %w", node, err)
|
||||
}
|
||||
store, err := js.ObjectStore(working, bucket)
|
||||
if errors.Is(err, jetstream.ErrBucketNotFound) {
|
||||
return filesSweep{}, nil // a node never sent a reference has no bucket yet
|
||||
}
|
||||
if err != nil {
|
||||
return filesSweep{}, fmt.Errorf("the bucket cannot be read: %w", err)
|
||||
}
|
||||
defer lockFilesOf(node)()
|
||||
latest, err := latestDeclarations(working, conn, []string{node})
|
||||
if err != nil {
|
||||
return filesSweep{}, err
|
||||
}
|
||||
named := map[string]bool{}
|
||||
if body, ok := latest[node]; ok {
|
||||
if named, err = namedByLatest(node, body); err != nil {
|
||||
return filesSweep{}, err
|
||||
}
|
||||
}
|
||||
return sweepDeclarationFiles(working, busSweptBucket{store}, named, time.Now())
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"reflect"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
)
|
||||
|
||||
// 1617/AC2 (unit): a new module whose first build a merge's hearing asked is not asked again when its batch is cut,
|
||||
// and one whose ask failed, was never handed over or is long lost is.
|
||||
func TestAModuleWhoseBuildIsComingIsNotAskedAgain(t *testing.T) {
|
||||
now := time.Date(2026, 10, 21, 12, 0, 0, 0, time.UTC)
|
||||
source := buildSource{Repository: "http://forge/novox/mesh-catalog.git", Seat: "git"}
|
||||
asked := func(path string, mod func(*inventory.RequestOutcome)) inventory.RequestOutcome {
|
||||
r := inventory.RequestOutcome{BuildRequest: inventory.BuildRequest{ID: "b-" + path, Repository: source.Repository,
|
||||
Seat: "git", Path: path, For: "merge", At: now.Add(-3 * time.Minute)}}
|
||||
if mod != nil {
|
||||
mod(&r)
|
||||
}
|
||||
return r
|
||||
}
|
||||
requests := []inventory.RequestOutcome{
|
||||
asked("modules/go", nil),
|
||||
asked("modules/failed", func(r *inventory.RequestOutcome) { r.Heard, r.Failed = true, "no module.json" }),
|
||||
asked("modules/notasked", func(r *inventory.RequestOutcome) { r.NotAsked = "the bus is silent" }),
|
||||
asked("modules/lost", func(r *inventory.RequestOutcome) { r.At = now.Add(-2 * time.Hour) }),
|
||||
asked("modules/byhand", func(r *inventory.RequestOutcome) { r.For = "build" }),
|
||||
}
|
||||
got := unaskedDirs([]string{"modules/go", "modules/failed", "modules/notasked", "modules/lost", "modules/byhand",
|
||||
"modules/nodejs"}, requests, source, now)
|
||||
want := []string{"modules/failed", "modules/notasked", "modules/lost", "modules/byhand", "modules/nodejs"}
|
||||
if !reflect.DeepEqual(got, want) {
|
||||
t.Fatalf("asked again %v, want %v", got, want)
|
||||
}
|
||||
}
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"context"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
@@ -99,3 +100,38 @@ func TestAMergeBuildsItsNewModuleBesideItsPlan(t *testing.T) {
|
||||
t.Fatalf("a merge adding no module asked for %v", *asked)
|
||||
}
|
||||
}
|
||||
|
||||
// 1617/AC1 (integration): a merge adding a module to a repository the mesh builds modules from, heard while a batch of
|
||||
// that repository is still assembling (its window running), asks the module's first build when it is heard, so the
|
||||
// module is registered while the batch waits; cutting the batch asks it no second time (1617/AC2).
|
||||
func TestAMergeAddingAModuleAsksItsBuildBeforeItsBatchIsCut(t *testing.T) {
|
||||
open := windowed(t)
|
||||
groupsWaiting(t, "pr/c1")
|
||||
asked := asksWithPaths(t)
|
||||
catalog := func(commit string, made time.Time, paths ...string) link.SourceMoved {
|
||||
return link.SourceMoved{Owner: "novox", Repo: "mesh-catalog", Base: "main", Commit: commit,
|
||||
Head: "pr/" + commit, MergedAt: made.Format(time.RFC3339Nano), Paths: paths}
|
||||
}
|
||||
hear(t, open, catalog("c1", t0, "modules/app/index.ts"), t0)
|
||||
if len(*asked) != 0 {
|
||||
t.Fatalf("a merge adding no module asked %v", *asked)
|
||||
}
|
||||
hear(t, open, lone(catalog("c2", t0.Add(10*time.Second), "modules/go/packages.json", "modules/go/module.json")), t0.Add(10*time.Second))
|
||||
if ws, bs := walks(t, open); len(ws) != 0 || len(bs) != 1 {
|
||||
t.Fatalf("the batch should still be assembling: %d walk(s), %d batch(es)", len(ws), len(bs))
|
||||
}
|
||||
want := [3]string{"novox/mesh-catalog", "modules/go", "main"}
|
||||
if len(*asked) != 1 || (*asked)[0] != want {
|
||||
t.Fatalf("heard with its batch assembling, the merge asked %v, not %v", *asked, want)
|
||||
}
|
||||
cutAt(t, open, t0.Add(10*time.Second+90*time.Second))
|
||||
n := 0
|
||||
for _, a := range *asked {
|
||||
if a[1] == "modules/go" {
|
||||
n++
|
||||
}
|
||||
}
|
||||
if n != 1 {
|
||||
t.Fatalf("modules/go was asked %d times: %v", n, *asked)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,198 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
)
|
||||
|
||||
// heldElsewhere is the plans held by another act for longer than any test runs: waiting for them ends only with the
|
||||
// context, and a look that does not wait is told at once that they are busy.
|
||||
func heldElsewhere(ctx context.Context, wait bool) (func(), error) {
|
||||
if !wait {
|
||||
return nil, inventory.ErrPlansBusy
|
||||
}
|
||||
<-ctx.Done()
|
||||
return nil, inventory.ErrPlansBusy
|
||||
}
|
||||
|
||||
// 1708/AC1: given the plans held by another act, when the event loop takes a build's outcome, then it returns at
|
||||
// once, without waiting the plans' two minutes (on 2026-10-21 at 14:20:45 nodered's outcome held the loop until
|
||||
// 14:22:45, and the operator was told the controller stopped listening).
|
||||
func TestAnOutcomeDoesNotWaitForThePlansAnotherActHolds(t *testing.T) {
|
||||
ctx, cancel := context.WithTimeout(t.Context(), 3*time.Second)
|
||||
defer cancel()
|
||||
start := time.Now()
|
||||
release, err := holdPlansForOutcome(ctx, heldElsewhere)
|
||||
if took := time.Since(start); took > time.Second {
|
||||
t.Fatalf("taking an outcome waited %s for plans another act holds; the event loop takes nothing meanwhile", took)
|
||||
}
|
||||
if err == nil || release != nil {
|
||||
t.Fatalf("plans another act holds were said to be held: release %v, err %v", release != nil, err)
|
||||
}
|
||||
}
|
||||
|
||||
// 1708/AC4: given the plans free, when the event loop takes a build's outcome, then it holds them and marks the plan
|
||||
// from the outcome directly, as before.
|
||||
func TestAnOutcomeHoldsThePlansWhenTheyAreFree(t *testing.T) {
|
||||
let := false
|
||||
release, err := holdPlansForOutcome(t.Context(), func(context.Context, bool) (func(), error) {
|
||||
return func() { let = true }, nil
|
||||
})
|
||||
if err != nil || release == nil {
|
||||
t.Fatalf("free plans were not held: %v", err)
|
||||
}
|
||||
release()
|
||||
if !let {
|
||||
t.Fatal("the hold given was not the one taken")
|
||||
}
|
||||
}
|
||||
|
||||
// withLooker installs a looker that is never run, and answers it, so a test can read whether a look was asked.
|
||||
func withLooker(t *testing.T) *looker {
|
||||
t.Helper()
|
||||
l := newLooker(func(context.Context) bool { return false })
|
||||
lookerMu.Lock()
|
||||
before := theLooker
|
||||
theLooker = l
|
||||
lookerMu.Unlock()
|
||||
t.Cleanup(func() {
|
||||
lookerMu.Lock()
|
||||
theLooker = before
|
||||
lookerMu.Unlock()
|
||||
})
|
||||
return l
|
||||
}
|
||||
|
||||
// 1708/AC2: given an outcome left to the build records because the plans were held, when it is left, then the look
|
||||
// at the plans is asked for, so the records settle it once the holder lets go rather than at the next event.
|
||||
func TestAnOutcomeLeftToTheRecordsAsksForTheLook(t *testing.T) {
|
||||
l := withLooker(t)
|
||||
leaveToRecords("nodered", "build-1708-ac2", "", "", inventory.ErrPlansBusy)
|
||||
select {
|
||||
case <-l.signal:
|
||||
default:
|
||||
t.Fatal("an outcome left to the build records asked for no look: the plan waits for whatever event comes next")
|
||||
}
|
||||
}
|
||||
|
||||
// 1708/AC3: given an outcome left to the records that said what the build did (ADR 0350 §4), when the look settles
|
||||
// the plan from the build records by that build's id, then the module keeps those words, once.
|
||||
func TestTheRecordsSettleAnOutcomeWithWhatTheBuildDid(t *testing.T) {
|
||||
withLooker(t)
|
||||
asked := time.Date(2026, 10, 21, 12, 15, 0, 0, time.UTC)
|
||||
plan := func() inventory.Plan {
|
||||
return inventory.Plan{ID: "plan-1708", Tiers: [][]string{{"nodered"}},
|
||||
Modules: map[string]*inventory.PlanModule{"nodered": {State: "asked", AskedAt: &asked, Build: "build-1708-ac3"}}}
|
||||
}
|
||||
byID := map[string]inventory.Build{"build-1708-ac3": {ID: "build-1708-ac3", Module: "nodered", Commit: "4efac366",
|
||||
At: asked.Add(5 * time.Minute)}}
|
||||
leaveToRecords("nodered", "build-1708-ac3", "reused from build-1708-check", "", inventory.ErrPlansBusy)
|
||||
|
||||
p := plan()
|
||||
if !settleFromRecords(&p, p.Tiers[0], nil, byID) || p.Modules["nodered"].State != "built" {
|
||||
t.Fatalf("the records did not settle the outcome: %+v", p.Modules["nodered"])
|
||||
}
|
||||
if got := p.Modules["nodered"].Made; got != "reused from build-1708-check" {
|
||||
t.Fatalf("settled from the records, the module says %q of what its build did, not the outcome's words", got)
|
||||
}
|
||||
again := plan()
|
||||
settleFromRecords(&again, again.Tiers[0], nil, byID)
|
||||
if got := again.Modules["nodered"].Made; got != "" {
|
||||
t.Fatalf("an outcome's words were kept for a second settle: %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
// 1708/AC5: given a serving controller (its look at the plans running), when an outcome was taken into the plans,
|
||||
// then moving them (a release's sends, which may retry for minutes) is left to the look, off the event loop
|
||||
// (2026-10-21 14:28:40 to 14:32:10: the loop sent okioki's release to g14 after mesh-delivery-builder's outcome).
|
||||
func TestAnOutcomeLeavesMovingThePlansToTheLook(t *testing.T) {
|
||||
l := withLooker(t)
|
||||
moved := false
|
||||
advanceAfterOutcome(func() { moved = true })
|
||||
if moved {
|
||||
t.Fatal("the plans were moved on the event loop after an outcome, where a send that retries holds every message")
|
||||
}
|
||||
select {
|
||||
case <-l.signal:
|
||||
default:
|
||||
t.Fatal("the look was not asked to move the plans after an outcome")
|
||||
}
|
||||
}
|
||||
|
||||
// 1708/AC5: given no look running (a command, which has no event loop), when an outcome was taken, then the plans are
|
||||
// moved at once, as before.
|
||||
func TestAnOutcomeMovesThePlansWhereNoLookRuns(t *testing.T) {
|
||||
lookerMu.Lock()
|
||||
before := theLooker
|
||||
theLooker = nil
|
||||
lookerMu.Unlock()
|
||||
t.Cleanup(func() {
|
||||
lookerMu.Lock()
|
||||
theLooker = before
|
||||
lookerMu.Unlock()
|
||||
})
|
||||
moved := false
|
||||
advanceAfterOutcome(func() { moved = true })
|
||||
if !moved {
|
||||
t.Fatal("with no look running, an outcome left the plans unmoved")
|
||||
}
|
||||
}
|
||||
|
||||
// 1708/AC6: given a registration the mesh refused (its build recorded as worked, the outcome failed), left to the
|
||||
// records because the plans were held, when the look settles the plan from the records, then the module and the plan
|
||||
// are failed, never built: settled as built, the tier would be sent the version from before (reviewer-1708@shanks).
|
||||
func TestARefusedRegistrationLeftToTheRecordsFailsThePlan(t *testing.T) {
|
||||
withLooker(t)
|
||||
asked := time.Date(2026, 10, 21, 12, 15, 0, 0, time.UTC)
|
||||
p := inventory.Plan{ID: "plan-1708-ac6", State: inventory.PlanBuilding, Tiers: [][]string{{"nodered"}},
|
||||
Modules: map[string]*inventory.PlanModule{"nodered": {State: "asked", AskedAt: &asked, Build: "build-1708-ac6"}}}
|
||||
byID := map[string]inventory.Build{"build-1708-ac6": {ID: "build-1708-ac6", Module: "nodered", Commit: "4efac366",
|
||||
At: asked.Add(5 * time.Minute)}}
|
||||
leaveToRecords("nodered", "build-1708-ac6", "", "the mesh does not register it: a port is taken", inventory.ErrPlansBusy)
|
||||
|
||||
overlayRefusals(nil, byID, refusalsKeptFor([]string{"build-1708-ac6"}))
|
||||
settleFromRecords(&p, p.Tiers[0], nil, byID)
|
||||
if s := p.Modules["nodered"]; s.State != "failed" || p.State != inventory.PlanFailed {
|
||||
t.Fatalf("a refused registration left to the records settled as %q, the plan %q", s.State, p.State)
|
||||
}
|
||||
if !strings.Contains(p.Modules["nodered"].Why, "a port is taken") {
|
||||
t.Fatalf("the refusal's words were lost: %q", p.Modules["nodered"].Why)
|
||||
}
|
||||
}
|
||||
|
||||
// 1708/AC6: a refusal kept in the store (a controller replaced before the settle) fails the record as one kept in
|
||||
// the process does.
|
||||
func TestARefusalFromTheStoreFailsTheRecord(t *testing.T) {
|
||||
recorded := map[string][]inventory.Build{"nodered": {{ID: "b-store", Module: "nodered"}, {ID: "b-other", Module: "nodered"}}}
|
||||
byID := map[string]inventory.Build{"b-store": {ID: "b-store", Module: "nodered"}}
|
||||
overlayRefusals(recorded, byID, map[string]string{"b-store": "not registered"})
|
||||
if recorded["nodered"][0].Worked() || byID["b-store"].Worked() {
|
||||
t.Fatal("a build whose registration was refused still reads as worked")
|
||||
}
|
||||
if !recorded["nodered"][1].Worked() {
|
||||
t.Fatal("a build nobody refused was failed")
|
||||
}
|
||||
}
|
||||
|
||||
// 1708/AC7: given the records settle a module as deleted at its source, or fail the plan, when the look settles, then
|
||||
// it does what taking the outcome directly does: the deleted module is named to be forgotten, and the plan's failure
|
||||
// is said (reviewer-1708@shanks).
|
||||
func TestSettlingFromTheRecordsDoesWhatTakingTheOutcomeDoes(t *testing.T) {
|
||||
p := inventory.Plan{State: inventory.PlanFailed, Modules: map[string]*inventory.PlanModule{
|
||||
"gone": {State: planDeleted}, "x": {State: "failed"}, "y": {State: "built"}}}
|
||||
before := map[string]string{"gone": "asked", "x": "asked", "y": "built"}
|
||||
deleted, failed := settledEffects(before, &p, inventory.PlanBuilding)
|
||||
if len(deleted) != 1 || deleted[0] != "gone" {
|
||||
t.Fatalf("the module the records found deleted is not forgotten: %v", deleted)
|
||||
}
|
||||
if !failed {
|
||||
t.Fatal("the plan the records failed is not said as failed")
|
||||
}
|
||||
if d, f := settledEffects(map[string]string{"gone": planDeleted}, &p, inventory.PlanFailed); len(d) != 0 || f {
|
||||
t.Fatalf("what was deleted or failed before the settle is acted on again: %v %v", d, f)
|
||||
}
|
||||
}
|
||||
@@ -195,6 +195,29 @@ var plainWordings = map[string]func(conditions.Observation) words{
|
||||
"there yet is sealed by the controller until the vault holds it.", m, m),
|
||||
Resolved: "Resolved: the vault gave " + m + " the core's credentials"}
|
||||
}),
|
||||
kindFilesNotPut: worded(func(o conditions.Observation) words {
|
||||
m := machineOr(o, "a machine")
|
||||
return words{Headline: "The big files for " + m + " cannot be put on the bus",
|
||||
Needs: "look at the bus's permissions for the controller; the details give the reason.",
|
||||
Explanation: fmt.Sprintf("The mesh could not put the large files of %s's instructions in their place on the bus, "+
|
||||
"send after send, so %s is sent them inside the instructions instead, which may make them too big.", m, m),
|
||||
Resolved: "Resolved: the big files for " + m + " reach the bus again"}
|
||||
}),
|
||||
kindDeclarationNearLimit: worded(func(o conditions.Observation) words {
|
||||
m := machineOr(o, "a machine")
|
||||
if o.Severity == conditions.Urgent {
|
||||
return words{Headline: m + " cannot be sent its instructions",
|
||||
Needs: "make the instructions for " + m + " smaller, for example by taking a module off it.",
|
||||
Explanation: fmt.Sprintf("The instructions for %s have grown past the largest message the bus takes, so "+
|
||||
"nothing new reaches %s until they are smaller.", m, m),
|
||||
Resolved: "Resolved: the instructions for " + m + " fit the bus again"}
|
||||
}
|
||||
return words{Headline: "The instructions for " + m + " are nearly too big",
|
||||
Needs: "do not add more to " + m + " until this is dealt with; the details give the size.",
|
||||
Explanation: fmt.Sprintf("The instructions for %s are close to the largest message the bus takes. "+
|
||||
"One more module or secret may stop anything new from reaching %s.", m, m),
|
||||
Resolved: "Resolved: the instructions for " + m + " have room again"}
|
||||
}),
|
||||
kindCoreDowngrade: worded(func(o conditions.Observation) words {
|
||||
m := machineOr(o, "a machine")
|
||||
component := idPart(o, 0)
|
||||
|
||||
@@ -136,6 +136,18 @@ func askToLook() {
|
||||
}
|
||||
}
|
||||
|
||||
// askedToLook asks the serving controller's look at the plans, and says whether there was one to ask.
|
||||
func askedToLook() bool {
|
||||
lookerMu.RLock()
|
||||
l := theLooker
|
||||
lookerMu.RUnlock()
|
||||
if l == nil {
|
||||
return false
|
||||
}
|
||||
l.ask()
|
||||
return true
|
||||
}
|
||||
|
||||
// startPlanLooks starts the look at the open plans, and asks for the first: the controller took the lease, and a
|
||||
// plan left by a replaced controller resumes from there (ADR 0406 §6). It replaces the rollout's 30 second ticker.
|
||||
// It answers a channel that is closed once the look has returned after the context ended, for a test that restores what
|
||||
|
||||
@@ -282,6 +282,8 @@ func serve(ctx context.Context) (err error) {
|
||||
// assignment.
|
||||
inv.SetProvisioningPlans(inv.SentProvisioningPlans)
|
||||
go mirroringRecords(ctx, inv, bus.Conn, bus)
|
||||
// And the bucket of the files that travel by digest, judged against what each machine's latest declaration names (ADR 0486 §3).
|
||||
go sweepingDeclarationFiles(ctx, inv, bus.Conn)
|
||||
// Every call carries the lease's epoch, and its record is written only under the lease (novox/hq
|
||||
// to-be 45 §6).
|
||||
link.Calls.UnderLease(func() (uint64, error) { return theLease.epoch(ctx) })
|
||||
@@ -979,6 +981,18 @@ func deliver(ctx context.Context, d delivery, holder string, sending []readyNode
|
||||
return err
|
||||
}
|
||||
digest, err := d.declare(ctx, s, body)
|
||||
if errors.Is(err, errFilesNotPut) && s.declared.fileReferences {
|
||||
// **The grant that lets the controller write the bucket travels in the very declaration of the machine holding the
|
||||
// bus** (novox/hq ADR 0486 §2): until the bus has it, a put is refused, and the declaration that would deliver the
|
||||
// grant would wait for it. So that send, and only that, goes with every content inline, as to a node that reads no
|
||||
// references, and says so; the next one tries the bucket again.
|
||||
fmt.Printf("%s: the files that travel by digest could not be put in the bucket (%v): sending them inline\n", s.node, err)
|
||||
s.declared.fileReferences = false
|
||||
if body, err = s.declared.Body(); err != nil {
|
||||
return err
|
||||
}
|
||||
digest, err = d.declare(ctx, s, body)
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -1040,6 +1054,35 @@ type overTheBus struct {
|
||||
indent string
|
||||
}
|
||||
|
||||
// putFiles puts the contents s names by digest in the bucket.
|
||||
func (b overTheBus) putFiles(ctx context.Context, s readyNode) error {
|
||||
objects, err := s.declared.objects()
|
||||
if err != nil {
|
||||
return fmt.Errorf("%s: the files that travel by digest cannot be read: %w", s.node, err)
|
||||
}
|
||||
if len(objects) == 0 {
|
||||
return nil
|
||||
}
|
||||
bus, ok := b.server.Bus().(link.OverNATS)
|
||||
if !ok {
|
||||
return fmt.Errorf("%s: the declaration names files by digest and this bus has no object bucket", s.node)
|
||||
}
|
||||
putting, cancel := context.WithTimeout(ctx, 30*time.Second)
|
||||
defer cancel()
|
||||
store, err := openBucket(putting, bus.Conn, s.node)
|
||||
if err == nil {
|
||||
err = putObjects(putting, store, objects)
|
||||
}
|
||||
notePut(ctx, s.node, err)
|
||||
if err != nil {
|
||||
return fmt.Errorf("%s was not sent its declaration: %w: %w", s.node, errFilesNotPut, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// errFilesNotPut marks a send that stopped because the files it names by digest could not be put in the bucket.
|
||||
var errFilesNotPut = errors.New("the files it names by digest are not in the bucket")
|
||||
|
||||
func (b overTheBus) grant(ctx context.Context, sending []readyNode) error {
|
||||
// The bus's objects first, which every declaration implies (novox/hq issue 208): said and raised
|
||||
// when they cannot be, and never what holds the send back (assertOnSend).
|
||||
@@ -1052,9 +1095,20 @@ func (b overTheBus) grant(ctx context.Context, sending []readyNode) error {
|
||||
}
|
||||
|
||||
func (b overTheBus) declare(ctx context.Context, s readyNode, body []byte) (string, error) {
|
||||
if err := link.Declare(ctx, b.server.Bus(), b.signer, s.node, body, 15*time.Second); err != nil {
|
||||
// The contents the declaration names by digest are in the bucket before it is sent (novox/hq ADR 0486 §1, issue 1696): a node
|
||||
// that applies it finds every one. A node reading no references has none.
|
||||
// **Under the node's lock** from the put to the publication, so the sweep of its bucket cannot remove an object between them.
|
||||
unlock := lockFilesOf(s.node)
|
||||
err := b.putFiles(ctx, s)
|
||||
if err != nil {
|
||||
unlock()
|
||||
return "", err
|
||||
}
|
||||
declared := link.Declare(ctx, b.server.Bus(), b.signer, s.node, body, 15*time.Second)
|
||||
unlock()
|
||||
if declared != nil {
|
||||
return "", declared
|
||||
}
|
||||
// After it is away, not before. A digest recorded for something that failed to send would make
|
||||
// the machine look current for a declaration it never received.
|
||||
digest, err := recordSent(ctx, b.open.inventory, s.node, body, s.declared.Builds, s.declared.Epoch,
|
||||
|
||||
@@ -625,16 +625,124 @@ func planBuilt(ctx context.Context, open *stores, module, commit, failed string,
|
||||
planBuiltSaying(ctx, open, module, commit, failed, asked, id, "")
|
||||
}
|
||||
|
||||
// holdPlansForOutcome is how an outcome taken on the event loop holds the plans: **without waiting for them**
|
||||
// (novox/hq issue 1708). The loop takes one message at a time, and the plans' holder may keep them for minutes while
|
||||
// it sends machines their declarations; an outcome that waited its turn (up to inventory.HoldWaitFor, two minutes)
|
||||
// held every report, build and merge behind it for as long, and the watchdog said the controller stopped listening
|
||||
// (2026-10-21 14:20:45 to 14:22:45). Busy, the outcome is left to the build records, where it already is, and the
|
||||
// look settles the plan from them (leaveToRecords).
|
||||
func holdPlansForOutcome(ctx context.Context, hold func(context.Context, bool) (func(), error)) (func(), error) {
|
||||
return hold(ctx, false)
|
||||
}
|
||||
|
||||
// leaveToRecords says an outcome was not taken into the plans, which settle it from the build records (novox/hq issue
|
||||
// 214). When another act holds the plans it asks for the look, which waits for them off the event loop and settles
|
||||
// the plan from the records once they are let go (issue 1708); what the build did is kept for that settle by the
|
||||
// build's id, so the walk's module says it as it would have (ADR 0350 §4).
|
||||
func leaveToRecords(module, id, made, failed string, err error) {
|
||||
fmt.Printf("plans: %s's outcome is left to the build records: %v\n", module, err)
|
||||
if !errors.Is(err, inventory.ErrPlansBusy) {
|
||||
return
|
||||
}
|
||||
keepMadeFor(id, made, time.Now())
|
||||
// A refused outcome (failed, its build recorded as worked) is kept as refused, or the records settle it built.
|
||||
keepRefusalFor(id, failed, time.Now())
|
||||
askToLook()
|
||||
}
|
||||
|
||||
// refusalsKeptFor is the refusals kept in this process for these builds (keepRefusalFor), by id.
|
||||
func refusalsKeptFor(ids []string) map[string]string {
|
||||
out := map[string]string{}
|
||||
refusedLeftMu.Lock()
|
||||
defer refusedLeftMu.Unlock()
|
||||
for _, id := range ids {
|
||||
if k, ok := refusedLeft[id]; ok {
|
||||
out[id] = k.made
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// refusedLeft is why each outcome left to the records was refused, by the build's id: kept in the process beside the
|
||||
// store's record (RecordRefused), so a store that did not take it still fails the plan while this controller serves.
|
||||
var (
|
||||
refusedLeftMu sync.Mutex
|
||||
refusedLeft = map[string]madeKept{}
|
||||
)
|
||||
|
||||
// keepRefusalFor keeps why an outcome left to the records was refused, forgetting refusals older than madeLeftFor.
|
||||
func keepRefusalFor(id, why string, now time.Time) {
|
||||
if id == "" || why == "" {
|
||||
return
|
||||
}
|
||||
refusedLeftMu.Lock()
|
||||
defer refusedLeftMu.Unlock()
|
||||
for k, v := range refusedLeft {
|
||||
if now.Sub(v.at) > madeLeftFor {
|
||||
delete(refusedLeft, k)
|
||||
}
|
||||
}
|
||||
refusedLeft[id] = madeKept{made: why, at: now}
|
||||
}
|
||||
|
||||
// madeLeft is what each build left to the records did, by the build's id, until a settle takes it: kept in the
|
||||
// process, so a controller replaced before the settle loses only these words, never the outcome.
|
||||
var (
|
||||
madeLeftMu sync.Mutex
|
||||
madeLeft = map[string]madeKept{}
|
||||
)
|
||||
|
||||
type madeKept struct {
|
||||
made string
|
||||
at time.Time
|
||||
}
|
||||
|
||||
// madeLeftFor is how long words nobody settled are kept: a plan that ended or moved on never asks for them.
|
||||
const madeLeftFor = time.Hour
|
||||
|
||||
// keepMadeFor keeps what a build left to the records did for the settle that takes it, forgetting words older than
|
||||
// madeLeftFor.
|
||||
func keepMadeFor(id, made string, now time.Time) {
|
||||
if id == "" || made == "" {
|
||||
return
|
||||
}
|
||||
madeLeftMu.Lock()
|
||||
defer madeLeftMu.Unlock()
|
||||
for k, v := range madeLeft {
|
||||
if now.Sub(v.at) > madeLeftFor {
|
||||
delete(madeLeft, k)
|
||||
}
|
||||
}
|
||||
madeLeft[id] = madeKept{made: made, at: now}
|
||||
}
|
||||
|
||||
// takeMadeOf is what the build of this id did, if its outcome was left to the records with words; taken once.
|
||||
func takeMadeOf(id string) string {
|
||||
madeLeftMu.Lock()
|
||||
defer madeLeftMu.Unlock()
|
||||
k, ok := madeLeft[id]
|
||||
if !ok {
|
||||
return ""
|
||||
}
|
||||
delete(madeLeft, id)
|
||||
return k.made
|
||||
}
|
||||
|
||||
// planBuiltSaying is planBuilt keeping what the build did — "reused from <check build>" or "built: <why>" (novox/hq
|
||||
// ADR 0350 §4) — on the module of the walk it answers.
|
||||
func planBuiltSaying(ctx context.Context, open *stores, module, commit, failed string, asked time.Time, id, made string) {
|
||||
inv := open.inventory
|
||||
// One controller works the plans at a time (novox/hq issue 213); an outcome waits its turn rather
|
||||
// than write over what the holder is about to save. Not taken, it is still in the build records,
|
||||
// which the holder settles the plan from (issue 214).
|
||||
release, err := inv.HoldPlans(ctx, true)
|
||||
// One controller works the plans at a time (novox/hq issue 213). Not taken, the outcome is still in the build
|
||||
// records, which the holder settles the plan from (issue 214).
|
||||
release, err := holdPlansForOutcome(ctx, inv.HoldPlans)
|
||||
if err != nil {
|
||||
fmt.Printf("plans: %s's outcome is left to the build records: %v\n", module, err)
|
||||
if failed != "" && id != "" && errors.Is(err, inventory.ErrPlansBusy) {
|
||||
// In the store too, so a controller replaced before the look settles still fails the plan (issue 1708).
|
||||
if err := inv.RecordRefused(ctx, id, module, failed); err != nil {
|
||||
fmt.Printf("plans: %s's refusal could not be kept in the store, only in this controller: %v\n", module, err)
|
||||
}
|
||||
}
|
||||
leaveToRecords(module, id, made, failed, err)
|
||||
return
|
||||
}
|
||||
defer release()
|
||||
@@ -696,7 +804,19 @@ func planBuiltSaying(ctx context.Context, open *stores, module, commit, failed s
|
||||
fmt.Printf("%s: %s; the tiers after it are not asked\n", p.ID, p.Note)
|
||||
}
|
||||
}
|
||||
advanceHeld(ctx, open)
|
||||
advanceAfterOutcome(func() { advanceHeld(ctx, open) })
|
||||
}
|
||||
|
||||
// advanceAfterOutcome moves the plans after an outcome was taken into them: **in the serving controller, by asking
|
||||
// the look** (novox/hq issue 1708). The outcome is taken on the event loop, and moving the plans may send machines
|
||||
// their declarations, a release's send retried for minutes among them; done here it held every report, build and
|
||||
// merge behind it (2026-10-21 14:28:40 to 14:32:10). The look runs off the loop, holding the plans itself. Where no
|
||||
// look runs (a command, which has no event loop) the plans are moved at once, as before.
|
||||
func advanceAfterOutcome(advance func()) {
|
||||
if askedToLook() {
|
||||
return
|
||||
}
|
||||
advance()
|
||||
}
|
||||
|
||||
// startedBeside is the id of a started walk open beside this one that it is related to — a merge's walk past
|
||||
@@ -921,7 +1041,17 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan,
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
before, was := statesOf(p), p.State
|
||||
if settleFromRecords(p, tier, recorded, byID) {
|
||||
// As taking the outcome directly does (planBuiltSaying): a module deleted at its source is forgotten, and a
|
||||
// failed plan says what it built and never sent (issue 1708).
|
||||
deleted, failed := settledEffects(before, p, was)
|
||||
for _, m := range deleted {
|
||||
forgetDeleted(ctx, inv, m, p)
|
||||
}
|
||||
if failed {
|
||||
sayUnsent(p, rollsOut)
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
// Asked: wait for every build.
|
||||
@@ -2479,9 +2609,81 @@ func recordsOfAsked(ctx context.Context, inv *inventory.Inventory, p *inventory.
|
||||
}
|
||||
}
|
||||
}
|
||||
refused, err := refusalsOf(ctx, inv, recorded, byID)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
overlayRefusals(recorded, byID, refused)
|
||||
return recorded, byID, nil
|
||||
}
|
||||
|
||||
// overlayRefusals marks failed every build record whose registration the mesh refused (novox/hq issue 1708): the record
|
||||
// keeps the builder's words, which worked, and settled from it as it is the refused build would be built and sent.
|
||||
func overlayRefusals(recorded map[string][]inventory.Build, byID map[string]inventory.Build, refused map[string]string) {
|
||||
if len(refused) == 0 {
|
||||
return
|
||||
}
|
||||
fail := func(b *inventory.Build) {
|
||||
if why, ok := refused[b.ID]; ok && b.Worked() {
|
||||
b.Failed = why
|
||||
}
|
||||
}
|
||||
for m := range recorded {
|
||||
for i := range recorded[m] {
|
||||
fail(&recorded[m][i])
|
||||
}
|
||||
}
|
||||
for id, b := range byID {
|
||||
fail(&b)
|
||||
byID[id] = b
|
||||
}
|
||||
}
|
||||
|
||||
// refusalsOf is every refusal kept for the builds these records name, in the store and in this process.
|
||||
func refusalsOf(ctx context.Context, inv *inventory.Inventory, recorded map[string][]inventory.Build,
|
||||
byID map[string]inventory.Build) (map[string]string, error) {
|
||||
var ids []string
|
||||
for _, bs := range recorded {
|
||||
for _, b := range bs {
|
||||
ids = append(ids, b.ID)
|
||||
}
|
||||
}
|
||||
for id := range byID {
|
||||
ids = append(ids, id)
|
||||
}
|
||||
refused, err := inv.RefusedOf(ctx, ids)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the refused registrations cannot be read: %w", err)
|
||||
}
|
||||
for id, why := range refusalsKeptFor(ids) {
|
||||
refused[id] = why
|
||||
}
|
||||
return refused, nil
|
||||
}
|
||||
|
||||
// settledEffects is what settling from the records did that needs more than the plan: the modules it found deleted
|
||||
// at their source, and whether it failed the plan.
|
||||
func settledEffects(before map[string]string, p *inventory.Plan, was string) (deleted []string, failed bool) {
|
||||
for m, s := range p.Modules {
|
||||
if s != nil && s.State == planDeleted && before[m] != planDeleted {
|
||||
deleted = append(deleted, m)
|
||||
}
|
||||
}
|
||||
sort.Strings(deleted)
|
||||
return deleted, p.State == inventory.PlanFailed && was != inventory.PlanFailed
|
||||
}
|
||||
|
||||
// statesOf is each module's state in a plan, for settledEffects.
|
||||
func statesOf(p *inventory.Plan) map[string]string {
|
||||
out := map[string]string{}
|
||||
for m, s := range p.Modules {
|
||||
if s != nil {
|
||||
out[m] = s.State
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// settleFromRecords marks every module of the tier still `asked` built — or failed — from a build
|
||||
// recorded after it was asked, and says whether it changed anything (novox/hq 04-ISSUES/214).
|
||||
// Newest first, as Builds answers: the first record after the ask is the outcome of that ask.
|
||||
@@ -2531,6 +2733,10 @@ func settleFromRecords(p *inventory.Plan, tier []string, recorded map[string][]i
|
||||
s.State = "built"
|
||||
s.BuiltAt = &at
|
||||
s.Commit = outcome.Commit
|
||||
// What the build did, when its outcome was left to the records with words (issue 1708).
|
||||
if made := takeMadeOf(outcome.ID); made != "" {
|
||||
s.Made = made
|
||||
}
|
||||
} else {
|
||||
s.State = "failed"
|
||||
s.Why = outcome.Failed
|
||||
|
||||
@@ -39,6 +39,10 @@ type sendable struct {
|
||||
// ReadsModules says the machine's node-engine reads a module per resource and a build per module (novox/hq issue
|
||||
// 1145): only then are they sent, because an older node-engine refuses a key it does not know, whole.
|
||||
ReadsModules bool
|
||||
// fileReferences says the machine's node-engine reports the capability of reading a file given as a digest and a size
|
||||
// (declaration.FileReferencesCapability, novox/hq ADR 0486 §5): then every file over 64 KiB that may travel so does (Body,
|
||||
// objects). Not on the wire as a field; the bodies differ.
|
||||
fileReferences bool
|
||||
// moduleOf is the composer's owner of each resource id, and buildIDs the build each module carries (its id):
|
||||
// kept beside the resources, and on the wire only to a machine that reads them.
|
||||
moduleOf map[string]string
|
||||
@@ -115,12 +119,16 @@ type adoptionEnvelope struct {
|
||||
|
||||
// Body is the declaration's bytes, as sent and as digested.
|
||||
func (s sendable) Body() ([]byte, error) {
|
||||
envelope := map[string]any{"declaration": 1, "resources": s.Resources}
|
||||
resources, _, err := s.byReference()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
envelope := map[string]any{"declaration": 1, "resources": resources}
|
||||
if s.ReadsModules {
|
||||
// The module of each resource the composer owns (what the mesh adds of its own has none), and the build of each
|
||||
// module (novox/hq issue 1145): to a machine that said it reads them, and no other.
|
||||
named := make([]map[string]any, len(s.Resources))
|
||||
for n, r := range s.Resources {
|
||||
named := make([]map[string]any, len(resources))
|
||||
for n, r := range resources {
|
||||
module, owned := s.moduleOf[fmt.Sprint(r["id"])]
|
||||
if !owned {
|
||||
named[n] = r
|
||||
@@ -165,3 +173,19 @@ func (s sendable) Body() ([]byte, error) {
|
||||
}
|
||||
return json.Marshal(envelope)
|
||||
}
|
||||
|
||||
// byReference is the resources as the machine is sent them: with the contents of its large files replaced by their digests when
|
||||
// it reports it reads them (issue 1696), and the contents it replaced, by the name of the object that holds each.
|
||||
func (s sendable) byReference() ([]map[string]any, map[string][]byte, error) {
|
||||
if !s.fileReferences {
|
||||
return s.Resources, nil, nil
|
||||
}
|
||||
return referencing(s.Resources)
|
||||
}
|
||||
|
||||
// objects are the contents the body names by digest, by the name of each one's object: what is put in the bucket before the
|
||||
// body is sent. Empty for a machine that is sent every content inline.
|
||||
func (s sendable) objects() (map[string][]byte, error) {
|
||||
_, objects, err := s.byReference()
|
||||
return objects, err
|
||||
}
|
||||
|
||||
@@ -415,7 +415,59 @@ func askNewModules(ctx context.Context, inv *inventory.Inventory, m link.SourceM
|
||||
if len(added) == 0 || len(from) == 0 {
|
||||
return nil
|
||||
}
|
||||
return askBuilds(ctx, inv, m, buildSource{Repository: from[0].Source.Repository, Seat: from[0].Source.Seat}, added)
|
||||
source := buildSource{Repository: from[0].Source.Repository, Seat: from[0].Source.Seat}
|
||||
// Asked once (novox/hq issue 1617): a module is asked when its merge is heard, and again here when the batch is cut.
|
||||
var requests []inventory.RequestOutcome
|
||||
for _, dir := range added {
|
||||
if got, err := inv.RequestsNamed(ctx, inventory.BuildRequest{Repository: source.Repository, Path: dir}.Name()); err == nil {
|
||||
requests = append(requests, got...)
|
||||
}
|
||||
}
|
||||
return askBuilds(ctx, inv, m, source, unaskedDirs(added, requests, source, time.Now()))
|
||||
}
|
||||
|
||||
// askedInFlightFor is how long a merge's build request stands as the build coming for its module: past it, a lost one
|
||||
// is asked again.
|
||||
const askedInFlightFor = time.Hour
|
||||
|
||||
// unaskedDirs are the directories of a new module no build of a merge is on its way for: a kept request of the same
|
||||
// repository and directory, asked for a merge and handed over, whose outcome has not come and is not given up on,
|
||||
// is the build coming (novox/hq issue 1617).
|
||||
func unaskedDirs(added []string, requests []inventory.RequestOutcome, source buildSource, now time.Time) []string {
|
||||
var out []string
|
||||
for _, dir := range added {
|
||||
path := dir
|
||||
if path == "." {
|
||||
path = ""
|
||||
}
|
||||
coming := false
|
||||
for _, r := range requests {
|
||||
if r.For == "merge" && r.Repository == source.Repository && strings.Trim(r.Path, "/") == path &&
|
||||
r.NotAsked == "" && r.OutcomeUnknown == "" && !r.Heard && now.Sub(r.At) < askedInFlightFor {
|
||||
coming = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !coming {
|
||||
out = append(out, dir)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// askNewModulesAtHearing asks the first build of the modules a merge adds to a repository the mesh builds modules from,
|
||||
// when the merge is heard, not when its batch is cut (novox/hq issue 1617): a batch waits out its window and the walk
|
||||
// before it, and a module added by a merge delivered meanwhile could not be assigned. Outside the batch and its plan.
|
||||
func askNewModulesAtHearing(ctx context.Context, inv *inventory.Inventory, m link.SourceMoved, entries []inventory.Entry,
|
||||
read map[string][]inventory.ReadRepository, from []inventory.Entry) {
|
||||
if len(from) == 0 {
|
||||
return
|
||||
}
|
||||
_, added, _ := touchedBy(from, entries, m, read)
|
||||
if asked := askNewModules(ctx, inv, m, from, added); len(asked) > 0 {
|
||||
fmt.Printf("%s/%s merged into %s (%.8s) adds %s: the first build is asked now, before its batch is cut\n",
|
||||
m.Owner, m.Repo, m.Base, m.Commit, strings.Join(asked, ", "))
|
||||
}
|
||||
}
|
||||
|
||||
// askBuilds asks the first build of each directory of a merge's repository, as askNewModules does, from the source given.
|
||||
|
||||
@@ -3,7 +3,7 @@ module github.com/novox/mesh-controller
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021093645-465beb4ea5e3
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021134822-86f9a1a5c508
|
||||
github.com/jackc/pgx/v5 v5.10.0
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
|
||||
@@ -2,6 +2,10 @@ git.novox.be/novox/mesh-host v0.0.0-20261018171808-9d9e3a46c4f5 h1:E5stfIeCN6uYg
|
||||
git.novox.be/novox/mesh-host v0.0.0-20261018171808-9d9e3a46c4f5/go.mod h1:4wyoCr6IIIIarEUuROThWDj7DAvJvIidqMjWcpSnOwE=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021093645-465beb4ea5e3 h1:OKxU64kHN1yQ/GOa6LwoUwVmiQ74d83WMSZJRFPVm9c=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021093645-465beb4ea5e3/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021124357-863a981e014e h1:eq29ENikw3XfEXjxL5OdnTbavv5pu1/UnQhWWuf5V5I=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021124357-863a981e014e/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021134822-86f9a1a5c508 h1:nTQN4W6gfIUR0fLpA1IYb06TGr3MtuW0xtznjbcMGLM=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021134822-86f9a1a5c508/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
|
||||
@@ -0,0 +1,123 @@
|
||||
package broker
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/declaration"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
)
|
||||
|
||||
// The object buckets a declaration's large files travel in (novox/hq ADR 0486, issues 1696 and 1697): **one bucket for each
|
||||
// node**, `declaration-files-<node>`.
|
||||
//
|
||||
// **The controller alone writes them, and a node-engine reads an object by its name from its own bucket** (ADR 0486 §2): its
|
||||
// account may neither list the bucket nor write it, no account may touch another node's, and no other account has any access.
|
||||
// Why a bucket each: a subject pattern cannot tell a name from a wildcard. A grant of `...M.*` matches a publish to the literal
|
||||
// subject `...M.*` too (probed on nats-server 2.11.17 by the review of this issue: a node created a consumer filtered on
|
||||
// `$O.<bucket>.C.*` and read every object of the bucket), so in one shared bucket a node could read every other node's files.
|
||||
// With a bucket each, the same trick reads only the node's own files, which are the files of its own declaration.
|
||||
//
|
||||
// A bucket of the bus's object store is a stream, OBJ_<bucket>, whose messages are the objects' metadata
|
||||
// (`$O.<bucket>.M.<base64 of the name>`) and chunks (`$O.<bucket>.C.<nuid>`).
|
||||
|
||||
// DeclarationFilesBucket is the prefix of the buckets' names, which mesh-sdk fixes for the node-engine and the controller alike.
|
||||
const DeclarationFilesBucket = declaration.FileBucketPrefix
|
||||
|
||||
// DeclarationFilesBucketOf is the bucket a node's files are in: mesh-sdk's FileBucketFor, the one place the name is composed. A
|
||||
// node name it refuses (one that could carry a subject wildcard or separator into a grant) is an error, never a bucket.
|
||||
func DeclarationFilesBucketOf(node string) (string, error) { return declaration.FileBucketFor(node) }
|
||||
|
||||
// DeclarationFilesStreamOf is the stream under a node's bucket: mesh-sdk's FileStreamFor.
|
||||
func DeclarationFilesStreamOf(node string) (string, error) { return declaration.FileStreamFor(node) }
|
||||
|
||||
// DeclarationFilesMaxBytes is a bucket's size: the objects are immutable and regenerable, and one node's hold well under a hundred
|
||||
// megabytes.
|
||||
const DeclarationFilesMaxBytes = 1 << 30
|
||||
|
||||
// DeclarationFilesKeptFor is how long an object stays after no node's latest declaration names it (ADR 0486 §3).
|
||||
const DeclarationFilesKeptFor = 24 * time.Hour
|
||||
|
||||
// DeclarationFilesWrites is the one pattern the controller may publish to write a node's bucket: every object's metadata and
|
||||
// chunks. The server's API for the stream is the controller's `$JS.API.>`.
|
||||
func DeclarationFilesWrites(node string) (string, error) {
|
||||
bucket, err := DeclarationFilesBucketOf(node)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return "$O." + bucket + ".>", nil
|
||||
}
|
||||
|
||||
// DeclarationFilesReads is what a node-engine's account may publish to read an object of its own bucket by its name, and nothing
|
||||
// more. The set is the one node-engine's own call (mesh-host internal/link FileObject) makes, for B the node's bucket and S its
|
||||
// stream (issue 1697, notes 4 and 5):
|
||||
//
|
||||
// - the object's metadata, by a direct get of the last message of the one subject that names it. **Only the suffixed
|
||||
// form**: the bare `$JS.API.DIRECT.GET.S` takes a sequence number and would read any message of the stream.
|
||||
// - the ordered consumer on that object's chunks: `$JS.API.CONSUMER.CREATE.S.<name>.$O.B.C.<nuid>`. **Not `S.>`**:
|
||||
// a consumer filtered on `$O.B.M.>` would read every object's name.
|
||||
// - the flow-control replies of that consumer, without which a large object stalls at the flow-control window.
|
||||
// - the consumer's deletion, best effort: the consumer's inactivity threshold removes it if this is refused.
|
||||
//
|
||||
// **Never granted:** STREAM.INFO (a `subjects_filter` of `>` lists every name), STREAM.MSG.GET (any message by its sequence), a
|
||||
// publish to `$O.B.>` (a write), and any subject of another node's stream. A literal `*` in a subject matches the grant's `*`: it
|
||||
// reaches the objects of this node's own bucket and no other (above).
|
||||
func DeclarationFilesReads(node string) ([]string, error) {
|
||||
bucket, err := DeclarationFilesBucketOf(node)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
stream, err := DeclarationFilesStreamOf(node)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return []string{
|
||||
"$JS.API.DIRECT.GET." + stream + ".$O." + bucket + ".M.*",
|
||||
"$JS.API.CONSUMER.CREATE." + stream + ".*.$O." + bucket + ".C.*",
|
||||
"$JS.API.CONSUMER.DELETE." + stream + ".*",
|
||||
"$JS.FC." + stream + ".>",
|
||||
}, nil
|
||||
}
|
||||
|
||||
// EnsureDeclarationFiles creates the buckets of these nodes if absent and brings their options to match. A bucket exists whenever the
|
||||
// bus does, so a node that is starting finds it (ADR 0486 §4). An update, never a delete.
|
||||
func (j *JetStream) EnsureDeclarationFiles(nodes []string) error {
|
||||
js, err := jetstream.New(j.conn)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
return EnsureDeclarationFiles(ctx, js, nodes...)
|
||||
}
|
||||
|
||||
// EnsureDeclarationFiles is the same on a JetStream the caller holds.
|
||||
func EnsureDeclarationFiles(ctx context.Context, js jetstream.JetStream, nodes ...string) error {
|
||||
for _, node := range nodes {
|
||||
cfg, err := DeclarationFilesConfig(node)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := js.CreateOrUpdateObjectStore(ctx, cfg); err != nil {
|
||||
return fmt.Errorf("asserting the object bucket %s: %w", cfg.Bucket, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// DeclarationFilesConfig is a node's bucket as the controller asserts it.
|
||||
func DeclarationFilesConfig(node string) (jetstream.ObjectStoreConfig, error) {
|
||||
bucket, err := DeclarationFilesBucketOf(node)
|
||||
if err != nil {
|
||||
return jetstream.ObjectStoreConfig{}, err
|
||||
}
|
||||
return jetstream.ObjectStoreConfig{
|
||||
Bucket: bucket,
|
||||
Description: "the contents of the files over 64 KiB of the declaration of " + node + ", each under the hex of its sha256 " +
|
||||
"digest (novox/hq ADR 0486): written by the controller alone, read by the node-engine of that node by name, removed " +
|
||||
"a day after the node's latest declaration no longer names it",
|
||||
MaxBytes: DeclarationFilesMaxBytes,
|
||||
Storage: jetstream.FileStorage,
|
||||
}, nil
|
||||
}
|
||||
@@ -0,0 +1,316 @@
|
||||
//go:build integration
|
||||
|
||||
package broker
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/base64"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats-server/v2/server"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
"golang.org/x/crypto/bcrypt"
|
||||
)
|
||||
|
||||
// nodeEnginesFileObject is the node-engine's own call (mesh-host internal/link files.go, OverNATS.FileObject), as the
|
||||
// engine makes it: the object's metadata by a direct get of one subject, then its chunks through an ordered consumer
|
||||
// bound to the stream. It is copied here so the grants are judged by the call that has to work, and its subjects are the
|
||||
// ones the grant names (issue 1697, notes 4 and 5); the node-engine's own integration test holds the other copy.
|
||||
func nodeEnginesFileObject(ctx context.Context, nc *nats.Conn, node, name string) ([]byte, error) {
|
||||
js, err := nc.JetStream(nats.MaxWait(10 * time.Second))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
stream, err := DeclarationFilesStreamOf(node)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
bucket, err := DeclarationFilesBucketOf(node)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
meta, err := js.GetLastMsg(stream, "$O."+bucket+".M."+base64.URLEncoding.EncodeToString([]byte(name)),
|
||||
nats.DirectGet(), nats.Context(ctx))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var info struct {
|
||||
NUID string `json:"nuid"`
|
||||
Size uint64 `json:"size"`
|
||||
}
|
||||
if err := json.Unmarshal(meta.Data, &info); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
sub, err := js.SubscribeSync("$O."+bucket+".C."+info.NUID, nats.OrderedConsumer(), nats.BindStream(stream))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer sub.Unsubscribe()
|
||||
body := make([]byte, 0, info.Size)
|
||||
for uint64(len(body)) < info.Size {
|
||||
msg, err := sub.NextMsgWithContext(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
body = append(body, msg.Data...)
|
||||
}
|
||||
return body, nil
|
||||
}
|
||||
|
||||
// **The buckets' grants on a real server, judged by the node-engine's own call** (novox/hq ADR 0486 §2, issue 1697, 1697/AC2): the
|
||||
// controller's account writes an object into each node's bucket, a node-engine's account reads one of its own bucket by its name,
|
||||
// a large object included, and is refused a list of the bucket (a stream info asking for every subject), a message get by
|
||||
// sequence, a direct get by sequence, a consumer on the names, and a write; it is refused every one of those and the read by name on
|
||||
// another node's bucket, **the literal-`*` forms included** (a published subject `...M.*` or `...C.*` matches the grant's `*`: on
|
||||
// its own bucket that reaches its own files, on another's it is refused); another module's account is refused all of them.
|
||||
func TestNatsANodeEngineReadsAnObjectOfItsOwnBucketByNameAndNothingElse(t *testing.T) { // 1697/AC2
|
||||
hash, err := bcrypt.GenerateFromPassword([]byte("pw"), bcrypt.MinCost)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
users, err := Users(Records{Nodes: []string{"alpha", "beta"}, Assigned: map[string][]Declared{
|
||||
"alpha": {{Module: RuntimeModule}, {Module: "other-module"}},
|
||||
}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for i := range users {
|
||||
users[i].PasswordHash = string(hash)
|
||||
}
|
||||
accounts, err := ComposeAccounts(users)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
conf := filepath.Join(t.TempDir(), "bus.conf")
|
||||
if err := os.WriteFile(conf, []byte("listen: 127.0.0.1:-1\njetstream { store_dir: "+
|
||||
`"`+t.TempDir()+`"`+" }\n"+accounts), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
opts, err := server.ProcessConfigFile(conf)
|
||||
if err != nil {
|
||||
t.Fatalf("the composed accounts do not parse: %v", err)
|
||||
}
|
||||
opts.NoLog, opts.NoSigs = true, true
|
||||
s, err := server.NewServer(opts)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
go s.Start()
|
||||
if !s.ReadyForConnections(10 * time.Second) {
|
||||
t.Fatal("the bus did not come up")
|
||||
}
|
||||
t.Cleanup(s.Shutdown)
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
|
||||
defer cancel()
|
||||
connect := func(user string) (*nats.Conn, chan error) {
|
||||
t.Helper()
|
||||
refused := make(chan error, 16)
|
||||
nc, err := nats.Connect(s.ClientURL(), nats.UserInfo(user, "pw"), nats.CustomInboxPrefix("_INBOX."+user),
|
||||
nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refused <- err }))
|
||||
if err != nil {
|
||||
t.Fatalf("%s cannot connect: %v", user, err)
|
||||
}
|
||||
t.Cleanup(nc.Close)
|
||||
return nc, refused
|
||||
}
|
||||
|
||||
// The controller makes the buckets (as it does at its start) and writes into each: for alpha a small object and one of
|
||||
// several chunks, which needs the consumer's flow control; for beta one that holds a marker alpha must never read.
|
||||
controller, _ := connect(ControllerName)
|
||||
cjs, err := jetstream.New(controller)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := EnsureDeclarationFiles(ctx, cjs, "alpha", "beta"); err != nil {
|
||||
t.Fatalf("the controller cannot make the buckets: %v", err)
|
||||
}
|
||||
if err := EnsureDeclarationFiles(ctx, cjs, "alpha", "beta"); err != nil {
|
||||
t.Fatalf("asserting them again: %v", err)
|
||||
}
|
||||
put := func(node string, content []byte) string {
|
||||
t.Helper()
|
||||
store, err := cjs.ObjectStore(ctx, mustBucket(t, node))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
sum := sha256.Sum256(content)
|
||||
name := hex.EncodeToString(sum[:])
|
||||
if _, err := store.PutBytes(ctx, name, content); err != nil {
|
||||
t.Fatalf("the controller cannot write the object %s of %s: %v", name[:8], node, err)
|
||||
}
|
||||
return name
|
||||
}
|
||||
small := bytes.Repeat([]byte("small file\n"), 100)
|
||||
large := make([]byte, 900<<10)
|
||||
for i := range large {
|
||||
large[i] = byte(i*7 + i>>8)
|
||||
}
|
||||
secret := bytes.Repeat([]byte("BETA-SECRET "), 50)
|
||||
smallName, largeName, betaName := put("alpha", small), put("alpha", large), put("beta", secret)
|
||||
|
||||
alpha, alphaRefused := connect("node.alpha")
|
||||
for name, want := range map[string][]byte{smallName: small, largeName: large} {
|
||||
got, err := nodeEnginesFileObject(ctx, alpha, "alpha", name)
|
||||
if err != nil {
|
||||
t.Fatalf("a node-engine cannot read the object %s of its own bucket by its name: %v", name[:8], err)
|
||||
}
|
||||
if !bytes.Equal(got, want) {
|
||||
t.Fatalf("the node-engine read %d bytes of the object %s, want %d", len(got), name[:8], len(want))
|
||||
}
|
||||
}
|
||||
beta, _ := connect("node.beta")
|
||||
if got, err := nodeEnginesFileObject(ctx, beta, "beta", betaName); err != nil || !bytes.Equal(got, secret) {
|
||||
t.Fatalf("beta's node-engine reads its own object: %v", err)
|
||||
}
|
||||
select {
|
||||
case err := <-alphaRefused:
|
||||
t.Fatalf("a node-engine's own read was refused something: %v", err)
|
||||
default:
|
||||
}
|
||||
|
||||
// What each account is refused, judged by the server.
|
||||
alphaStream, alphaBucket := mustStream(t, "alpha"), mustBucket(t, "alpha")
|
||||
betaStream, betaBucket := mustStream(t, "beta"), mustBucket(t, "beta")
|
||||
listing := []byte(`{"subjects_filter":">"}`)
|
||||
bySeq := []byte(`{"seq":1}`)
|
||||
type refusal struct {
|
||||
what string
|
||||
do func(nc *nats.Conn)
|
||||
}
|
||||
onBucket := func(stream, bucket, which string) []refusal {
|
||||
return []refusal{
|
||||
{which + ": a list of the bucket (stream info asking for every subject)", func(nc *nats.Conn) {
|
||||
_, _ = nc.Request("$JS.API.STREAM.INFO."+stream, listing, 300*time.Millisecond)
|
||||
}},
|
||||
{which + ": a message get by sequence", func(nc *nats.Conn) {
|
||||
_, _ = nc.Request("$JS.API.STREAM.MSG.GET."+stream, bySeq, 300*time.Millisecond)
|
||||
}},
|
||||
{which + ": a direct get by sequence", func(nc *nats.Conn) {
|
||||
_, _ = nc.Request("$JS.API.DIRECT.GET."+stream, bySeq, 300*time.Millisecond)
|
||||
}},
|
||||
{which + ": a consumer on every object's name", func(nc *nats.Conn) {
|
||||
js, _ := nc.JetStream(nats.MaxWait(500 * time.Millisecond))
|
||||
if sub, err := js.SubscribeSync("$O."+bucket+".M.>", nats.OrderedConsumer(), nats.BindStream(stream)); err == nil {
|
||||
sub.Unsubscribe()
|
||||
}
|
||||
}},
|
||||
{which + ": a write of an object's metadata", func(nc *nats.Conn) {
|
||||
_ = nc.Publish("$O."+bucket+".M.abc", []byte(`{}`))
|
||||
_ = nc.Flush()
|
||||
}},
|
||||
{which + ": a write of a chunk", func(nc *nats.Conn) {
|
||||
_ = nc.Publish("$O."+bucket+".C.abc", []byte("x"))
|
||||
_ = nc.Flush()
|
||||
}},
|
||||
}
|
||||
}
|
||||
refusedWith := func(who string, nc *nats.Conn, ch chan error, r refusal) {
|
||||
t.Helper()
|
||||
for len(ch) > 0 {
|
||||
<-ch
|
||||
}
|
||||
r.do(nc)
|
||||
_ = nc.Flush()
|
||||
deadline := time.After(3 * time.Second)
|
||||
for {
|
||||
select {
|
||||
case err := <-ch:
|
||||
if errors.Is(err, nats.ErrPermissionViolation) {
|
||||
return
|
||||
}
|
||||
case <-deadline:
|
||||
t.Errorf("the server did not refuse %s %s", who, r.what)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
// Alpha's account on its own bucket (list, sequence reads, names, writes) and on beta's (all of that, and the read by name,
|
||||
// and the literal-`*` probes of the review).
|
||||
reachBeta := []refusal{
|
||||
{"beta's bucket: a read of one object by its name", func(nc *nats.Conn) {
|
||||
_, _ = nc.Request("$JS.API.DIRECT.GET."+betaStream+".$O."+betaBucket+".M."+
|
||||
base64.URLEncoding.EncodeToString([]byte(betaName)), nil, 300*time.Millisecond)
|
||||
}},
|
||||
{"beta's bucket: a direct get with a literal `*` for the name", func(nc *nats.Conn) {
|
||||
_, _ = nc.Request("$JS.API.DIRECT.GET."+betaStream+".$O."+betaBucket+".M.*", nil, 300*time.Millisecond)
|
||||
}},
|
||||
{"beta's bucket: a consumer with a literal `*` for the chunks", func(nc *nats.Conn) {
|
||||
js, _ := nc.JetStream(nats.MaxWait(500 * time.Millisecond))
|
||||
if sub, err := js.SubscribeSync("$O."+betaBucket+".C.*", nats.OrderedConsumer(), nats.BindStream(betaStream)); err == nil {
|
||||
sub.Unsubscribe()
|
||||
}
|
||||
}},
|
||||
}
|
||||
all := append(append(onBucket(alphaStream, alphaBucket, "its own bucket"), onBucket(betaStream, betaBucket, "beta's bucket")...), reachBeta...)
|
||||
for _, r := range all {
|
||||
nodeNC, nodeCh := connect("node.alpha")
|
||||
refusedWith("a node-engine", nodeNC, nodeCh, r)
|
||||
}
|
||||
// Another module is refused all of that on both buckets, and the reads by name on both.
|
||||
for _, r := range append(append(onBucket(alphaStream, alphaBucket, "alpha's bucket"), reachBeta...),
|
||||
refusal{"alpha's bucket: a read of one object by its name", func(nc *nats.Conn) {
|
||||
_, _ = nc.Request("$JS.API.DIRECT.GET."+alphaStream+".$O."+alphaBucket+".M."+
|
||||
base64.URLEncoding.EncodeToString([]byte(smallName)), nil, 300*time.Millisecond)
|
||||
}}) {
|
||||
otherNC, otherCh := connect("alpha.other-module")
|
||||
refusedWith("another module", otherNC, otherCh, r)
|
||||
}
|
||||
|
||||
// The literal `*` on alpha's own bucket reaches alpha's own files and never beta's: the chunks it is given hold no marker of beta's.
|
||||
js, err := alpha.JetStream(nats.MaxWait(2 * time.Second))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if sub, err := js.SubscribeSync("$O."+alphaBucket+".C.*", nats.OrderedConsumer(), nats.BindStream(alphaStream)); err == nil {
|
||||
for i := 0; i < 20; i++ {
|
||||
msg, err := sub.NextMsg(300 * time.Millisecond)
|
||||
if err != nil {
|
||||
break
|
||||
}
|
||||
if bytes.Contains(msg.Data, []byte("BETA-SECRET")) {
|
||||
t.Fatal("a literal `*` on alpha's bucket gave it a chunk of beta's file")
|
||||
}
|
||||
}
|
||||
sub.Unsubscribe()
|
||||
}
|
||||
|
||||
// Nothing was written by the refused: each bucket holds what the controller wrote.
|
||||
for node, want := range map[string]int{"alpha": 2, "beta": 1} {
|
||||
store, err := cjs.ObjectStore(ctx, mustBucket(t, node))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
infos, err := store.List(ctx)
|
||||
if err != nil || len(infos) != want {
|
||||
t.Errorf("the bucket of %s holds %d objects (%v), want the %d the controller wrote", node, len(infos), err, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func mustBucket(t *testing.T, node string) string {
|
||||
t.Helper()
|
||||
b, err := DeclarationFilesBucketOf(node)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
||||
func mustStream(t *testing.T, node string) string {
|
||||
t.Helper()
|
||||
b, err := DeclarationFilesStreamOf(node)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return b
|
||||
}
|
||||
@@ -0,0 +1,131 @@
|
||||
package broker
|
||||
|
||||
import (
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// 1697/AC1: the composed accounts give a node-engine exactly the four reads of one object by its name, on its own node's bucket and
|
||||
// no other's; the controller the write of every node's bucket; and no other principal anything of any bucket.
|
||||
func TestEachNodesBucketIsWrittenByTheControllerAndReadByNameByThatNodesEngine(t *testing.T) { // 1697/AC1
|
||||
node, err := PermissionsFor(Principal{Kind: KindNode, Node: "alpha", PasswordHash: "x"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var mine []string
|
||||
for _, s := range node.Publish {
|
||||
if strings.Contains(s, "declaration-files") {
|
||||
mine = append(mine, s)
|
||||
}
|
||||
}
|
||||
want := []string{
|
||||
"$JS.API.CONSUMER.CREATE.OBJ_declaration-files-alpha.*.$O.declaration-files-alpha.C.*",
|
||||
"$JS.API.CONSUMER.DELETE.OBJ_declaration-files-alpha.*",
|
||||
"$JS.API.DIRECT.GET.OBJ_declaration-files-alpha.$O.declaration-files-alpha.M.*",
|
||||
"$JS.FC.OBJ_declaration-files-alpha.>",
|
||||
}
|
||||
if !slices.Equal(mine, want) {
|
||||
t.Fatalf("a node-engine may publish %q of the buckets, want exactly %q", mine, want)
|
||||
}
|
||||
// What would make a list, a write, or a read of another node's files: none of it, in any pattern a node holds. (The real
|
||||
// server's judgement of the wildcard forms is the live test's.)
|
||||
for _, refused := range []string{
|
||||
"$JS.API.STREAM.INFO.OBJ_declaration-files-alpha",
|
||||
"$JS.API.STREAM.MSG.GET.OBJ_declaration-files-alpha",
|
||||
"$JS.API.DIRECT.GET.OBJ_declaration-files-alpha",
|
||||
"$JS.API.DIRECT.GET.OBJ_declaration-files-alpha.$O.declaration-files-alpha.M.abc.def",
|
||||
"$JS.API.CONSUMER.CREATE.OBJ_declaration-files-alpha.x.$O.declaration-files-alpha.M.YWJj",
|
||||
"$JS.API.CONSUMER.CREATE.OBJ_declaration-files-alpha.x.$O.declaration-files-alpha.M.YWJj.more",
|
||||
"$JS.API.CONSUMER.CREATE.OBJ_declaration-files-alpha",
|
||||
"$O.declaration-files-alpha.M.abc",
|
||||
"$O.declaration-files-alpha.C.abc",
|
||||
// another node's bucket, however it is named
|
||||
"$JS.API.DIRECT.GET.OBJ_declaration-files-beta.$O.declaration-files-beta.M.YWJj",
|
||||
"$JS.API.CONSUMER.CREATE.OBJ_declaration-files-beta.h4sh.$O.declaration-files-beta.C.NUID",
|
||||
"$JS.API.CONSUMER.DELETE.OBJ_declaration-files-beta.h4sh",
|
||||
"$JS.FC.OBJ_declaration-files-beta.x",
|
||||
// and the old shared bucket
|
||||
"$JS.API.DIRECT.GET.OBJ_declaration-files.$O.declaration-files.M.YWJj",
|
||||
} {
|
||||
if MayPublish(node, refused) {
|
||||
t.Errorf("a node-engine may publish %s, which lists or writes a bucket or reads another node's", refused)
|
||||
}
|
||||
}
|
||||
for _, kept := range []string{
|
||||
"$JS.API.DIRECT.GET.OBJ_declaration-files-alpha.$O.declaration-files-alpha.M.YWJj",
|
||||
"$JS.API.CONSUMER.CREATE.OBJ_declaration-files-alpha.h4sh.$O.declaration-files-alpha.C.NUID",
|
||||
} {
|
||||
if !MayPublish(node, kept) {
|
||||
t.Errorf("a node-engine may not publish %s, which reads one object of its own bucket by its name", kept)
|
||||
}
|
||||
}
|
||||
|
||||
controller, err := PermissionsFor(Principal{Kind: KindController, PasswordHash: "x", FilesFor: []string{"alpha", "beta"}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, s := range []string{"$O.declaration-files-alpha.M.abc", "$O.declaration-files-beta.C.nuid"} {
|
||||
if !MayPublish(controller, s) {
|
||||
t.Errorf("the controller may not write %s", s)
|
||||
}
|
||||
}
|
||||
if MayPublish(controller, "$O.declaration-files-gamma.M.abc") {
|
||||
t.Error("the controller may write the bucket of a node it was not given")
|
||||
}
|
||||
|
||||
for _, p := range []Principal{
|
||||
{Kind: KindModule, Module: "mesh-vault", PasswordHash: "x"},
|
||||
{Kind: KindNodeTools, Node: "alpha", PasswordHash: "x"},
|
||||
{Kind: KindPerson, Module: "jochen", Invokes: []string{"*"}, PasswordHash: "x"},
|
||||
} {
|
||||
perms, err := PermissionsFor(p)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, s := range append(append([]string(nil), perms.Publish...), perms.Subscribe...) {
|
||||
if strings.Contains(s, "declaration-files") {
|
||||
t.Errorf("%s holds %s, and no account but the controller's and the node-engines' has any access", p.Username(), s)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// The users the controller composes carry each node's bucket to the controller and to that node alone.
|
||||
func TestTheComposedUsersGiveEachNodeItsOwnBucketAndTheControllerAll(t *testing.T) {
|
||||
users, err := Users(Records{Nodes: []string{"alpha", "beta"}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var controller Principal
|
||||
for _, u := range users {
|
||||
if u.Kind == KindController {
|
||||
controller = u
|
||||
}
|
||||
}
|
||||
if !slices.Equal(controller.FilesFor, []string{"alpha", "beta"}) {
|
||||
t.Fatalf("the controller writes the buckets of %v", controller.FilesFor)
|
||||
}
|
||||
}
|
||||
|
||||
// A node name that could carry a subject wildcard or separator into a grant is refused with an error, never composed into a bucket
|
||||
// (ADR 0488): the grants of such a name would reach another node's bucket.
|
||||
func TestANodeNameThatCarriesAWildcardGetsNoBucketAndNoGrant(t *testing.T) {
|
||||
for _, node := range []string{"", "*", ">", "a.b", "a b", "a*"} {
|
||||
if b, err := DeclarationFilesBucketOf(node); err == nil {
|
||||
t.Errorf("node %q got the bucket %q", node, b)
|
||||
}
|
||||
if _, err := DeclarationFilesReads(node); err == nil {
|
||||
t.Errorf("node %q got read grants", node)
|
||||
}
|
||||
if _, err := DeclarationFilesWrites(node); err == nil {
|
||||
t.Errorf("node %q got write grants", node)
|
||||
}
|
||||
if _, err := DeclarationFilesConfig(node); err == nil {
|
||||
t.Errorf("node %q got a bucket config", node)
|
||||
}
|
||||
}
|
||||
if b, err := DeclarationFilesBucketOf("g14"); err != nil || b != "declaration-files-g14" {
|
||||
t.Errorf("g14: %q %v", b, err)
|
||||
}
|
||||
}
|
||||
@@ -104,6 +104,10 @@ type Principal struct {
|
||||
// Username and inbox are <node>.<module>.<process id>.
|
||||
Process string
|
||||
|
||||
// FilesFor are the nodes whose declaration-file buckets this principal writes, for the controller alone (novox/hq ADR 0486 §2):
|
||||
// every node of the mesh, one bucket each.
|
||||
FilesFor []string
|
||||
|
||||
Emits []string
|
||||
Consumes []string
|
||||
Serves []string
|
||||
@@ -736,6 +740,15 @@ func PermissionsFor(p Principal) (Permissions, error) {
|
||||
}
|
||||
pub = append(pub, "$KV."+bucket+".>")
|
||||
}
|
||||
// **And the object bucket a declaration's large files travel in** (novox/hq ADR 0486 §2, issue 1697): the
|
||||
// controller alone writes it, as a publish to the bucket's own subjects, which `$JS.API.>` does not cover.
|
||||
for _, node := range p.FilesFor {
|
||||
writes, err := DeclarationFilesWrites(node)
|
||||
if err != nil {
|
||||
return Permissions{}, err
|
||||
}
|
||||
pub = append(pub, writes)
|
||||
}
|
||||
// **And what the bus says about itself, read-only** (novox/hq to-be 45 §3, S9): a durable
|
||||
// consumer that gave up on a message, or one that was deleted. The server already publishes
|
||||
// both in the mesh's own account; the controller says each as a condition in the mesh's words.
|
||||
@@ -835,6 +848,13 @@ func PermissionsFor(p Principal) (Permissions, error) {
|
||||
return Permissions{}, err
|
||||
}
|
||||
pub = append(pub, haveKeys...)
|
||||
// And an object of the declaration bucket by its name, and nothing of the bucket's list or writing
|
||||
// (novox/hq ADR 0486 §2, issue 1697).
|
||||
fileReads, err := DeclarationFilesReads(p.Node)
|
||||
if err != nil {
|
||||
return Permissions{}, err
|
||||
}
|
||||
pub = append(pub, fileReads...)
|
||||
// And, while the vault serves it, the vault's move-in in its own name and nobody else's (issue 676): the
|
||||
// values sealed to this machine, which only it opens, handed in sealed to the vault's key for the call.
|
||||
if p.MovesIntoTheVault {
|
||||
|
||||
+1
-1
@@ -38,7 +38,7 @@ accounts {
|
||||
subscribe: { allow: ["_INBOX.enrol.one.>"] }
|
||||
} }
|
||||
{ user: "node.one", password: "$2a$11$nnnnnnnnnnnnnnnnnnnnnn", permissions: {
|
||||
publish: { allow: ["$JS.ACK.NODES.one.>", "$JS.API.CONSUMER.CREATE.KV_node-engine_have.*.$KV.node-engine_have.one.>", "$JS.API.CONSUMER.DELETE.KV_node-engine_have.>", "$JS.API.CONSUMER.INFO.NODES.one", "$JS.API.DIRECT.GET.KV_node-engine_have.$KV.node-engine_have.one.>", "$JS.API.STREAM.INFO.KV_node-engine_have", "$JS.FC.KV_node-engine_have.>", "$KV.node-engine_have.one.>", "$SRV.PING.node-tools.one", "mesh.control.one.>"] }
|
||||
publish: { allow: ["$JS.ACK.NODES.one.>", "$JS.API.CONSUMER.CREATE.KV_node-engine_have.*.$KV.node-engine_have.one.>", "$JS.API.CONSUMER.CREATE.OBJ_declaration-files-one.*.$O.declaration-files-one.C.*", "$JS.API.CONSUMER.DELETE.KV_node-engine_have.>", "$JS.API.CONSUMER.DELETE.OBJ_declaration-files-one.*", "$JS.API.CONSUMER.INFO.NODES.one", "$JS.API.DIRECT.GET.KV_node-engine_have.$KV.node-engine_have.one.>", "$JS.API.DIRECT.GET.OBJ_declaration-files-one.$O.declaration-files-one.M.*", "$JS.API.STREAM.INFO.KV_node-engine_have", "$JS.FC.KV_node-engine_have.>", "$JS.FC.OBJ_declaration-files-one.>", "$KV.node-engine_have.one.>", "$SRV.PING.node-tools.one", "mesh.control.one.>"] }
|
||||
subscribe: { allow: ["_DELIVER.one", "_DELIVER.one.>", "_INBOX.node.one.>", "mesh.node.one.ask.hand-over", "mesh.node.one.ask.judge-directories", "mesh.node.one.ask.move-secret", "mesh.node.one.ask.report", "mesh.node.one.ask.setuid-search", "mesh.node.one.declare"] }
|
||||
allow_responses: { max: 1, ttl: "1m" }
|
||||
} }
|
||||
|
||||
@@ -135,7 +135,7 @@ type Records struct {
|
||||
// correct.
|
||||
func Users(r Records) ([]Principal, error) {
|
||||
out := []Principal{{Kind: KindController, Uses: asksSeatOf(r), AsksIssued: issuedOf(r), AsksTheVault: vaultVerbsOf(r),
|
||||
AsksTheConfig: servedOf(r, VerbsTheControllerAsksTheConfig)}}
|
||||
AsksTheConfig: servedOf(r, VerbsTheControllerAsksTheConfig), FilesFor: sortedCopy(r.Nodes)}}
|
||||
|
||||
for _, node := range sortedCopy(r.Nodes) {
|
||||
witness := false
|
||||
|
||||
@@ -219,6 +219,13 @@ var WritersTable = []WriterRow{
|
||||
{State: "a machine's have", Writer: "the node-engine of the machine, under its own keys `<node>.>`",
|
||||
KeptIn: "key-value " + delivery.HaveBucket + ", twenty revisions of each key",
|
||||
Others: "read through `reads` as node-engine.have", Subjects: kvOf(delivery.HaveBucket), Writes: ownNodeKeys},
|
||||
// The contents of a declaration's files over 64 KiB (novox/hq ADR 0486 §2): the controller alone writes the bucket, each
|
||||
// object under the digest of its content; a node-engine reads one by its name, with no list and no write.
|
||||
{State: "a declaration's large files", Writer: "controller, each object under the sha256 digest of its content",
|
||||
KeptIn: "the object bucket " + DeclarationFilesBucket + "-<node> of each node, an object removed a day after the " +
|
||||
"node's latest declaration no longer names it",
|
||||
Others: "a node-engine reads an object of its own bucket by its name and may neither list the bucket nor write it",
|
||||
Subjects: []string{"$O.*.>"}, Writes: isController},
|
||||
{State: "the declared-config mirror", Writer: "controller (lease holder), from its store, only on a change",
|
||||
KeptIn: "key-value " + DeclaredConfigBucket, Others: "read through `reads` as mesh-controller." + MirrorDeclaredConfig,
|
||||
Subjects: kvOf(DeclaredConfigBucket), Writes: isController},
|
||||
|
||||
@@ -43,6 +43,8 @@ var designRows = []string{
|
||||
"the rollouts of the chain",
|
||||
// What each machine has of each module, written by its node-engine under its own keys (ADR 0429 §1, issue 1136).
|
||||
"a machine's have",
|
||||
// The contents of a declaration's large files, in the bus's object bucket (ADR 0486 §2, issue 1697).
|
||||
"a declaration's large files",
|
||||
"the declared-config mirror",
|
||||
"the predefined-keys mirror",
|
||||
// The provisioner client's word on what it placed (novox/hq ADR 0349 §4, design 57 P6).
|
||||
|
||||
@@ -100,8 +100,9 @@ var ErrPlansBusy = errors.New("another controller is working the plans")
|
||||
// returned function and by the session ending, so a controller that dies holding it holds nothing.
|
||||
//
|
||||
// wait false gives ErrPlansBusy at once when another holds them — the timer's way: the holder is
|
||||
// moving the plans already. wait true looks again every HoldPoll for up to HoldWaitFor — an
|
||||
// outcome's or a merge's way, which must be written.
|
||||
// moving the plans already; and a build outcome's on the event loop, which the build records keep
|
||||
// for the look to settle (novox/hq issue 1708). wait true looks again every HoldPoll for up to
|
||||
// HoldWaitFor — a merge's way, which must be written.
|
||||
func (i *Inventory) HoldPlans(ctx context.Context, wait bool) (func(), error) {
|
||||
deadline := time.Now().Add(HoldWaitFor)
|
||||
for {
|
||||
|
||||
@@ -0,0 +1,10 @@
|
||||
-- A build's outcome whose registration the mesh refused, by the build's id (novox/hq issue 1708). The build record says
|
||||
-- what the builder said, which worked; an outcome the event loop could not take into the plans (another act held them)
|
||||
-- is settled from the records by the look, and without this it would read the refused build as built. Kept in the
|
||||
-- store so a controller replaced before the settle still fails the plan.
|
||||
create table build_refused (
|
||||
id text primary key,
|
||||
module text not null,
|
||||
why text not null,
|
||||
at timestamptz not null default now()
|
||||
);
|
||||
@@ -0,0 +1,33 @@
|
||||
package inventory
|
||||
|
||||
import "context"
|
||||
|
||||
// RecordRefused keeps that the mesh refused to register a build's outcome, and why (novox/hq issue 1708): the build
|
||||
// record says the builder's words, which worked, and the look that settles a plan from the records reads this to fail
|
||||
// it instead. The first refusal of an id stands.
|
||||
func (i *Inventory) RecordRefused(ctx context.Context, id, module, why string) error {
|
||||
_, err := i.store.Pool().Exec(ctx,
|
||||
`insert into build_refused (id, module, why) values ($1, $2, $3) on conflict (id) do nothing`, id, module, why)
|
||||
return err
|
||||
}
|
||||
|
||||
// RefusedOf is why each of these builds' registration was refused, by id; a build not refused is absent.
|
||||
func (i *Inventory) RefusedOf(ctx context.Context, ids []string) (map[string]string, error) {
|
||||
out := map[string]string{}
|
||||
if len(ids) == 0 {
|
||||
return out, nil
|
||||
}
|
||||
rows, err := i.store.Pool().Query(ctx, `select id, why from build_refused where id = any($1)`, ids)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var id, why string
|
||||
if err := rows.Scan(&id, &why); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out[id] = why
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
//go:build integration
|
||||
|
||||
package inventory
|
||||
|
||||
import "testing"
|
||||
|
||||
// novox/hq issue 1708 (1708/AC6): a refused registration is kept by its build's id, the first refusal standing, and
|
||||
// read back for the builds asked about only.
|
||||
func TestARefusedRegistrationIsKeptByItsBuild(t *testing.T) {
|
||||
inv := ForTest(t)
|
||||
ctx := t.Context()
|
||||
if err := inv.RecordRefused(ctx, "build-1708-a", "nodered", "a port is taken"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := inv.RecordRefused(ctx, "build-1708-a", "nodered", "said again"); err != nil {
|
||||
t.Fatalf("a refusal said twice was refused: %v", err)
|
||||
}
|
||||
got, err := inv.RefusedOf(ctx, []string{"build-1708-a", "build-1708-b"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(got) != 1 || got["build-1708-a"] != "a port is taken" {
|
||||
t.Fatalf("the refusals read back are %v", got)
|
||||
}
|
||||
if none, err := inv.RefusedOf(ctx, nil); err != nil || len(none) != 0 {
|
||||
t.Fatalf("no builds asked about answered %v %v", none, err)
|
||||
}
|
||||
}
|
||||
@@ -184,6 +184,14 @@ func (b OverNATS) PublishMembershipAfter(ctx context.Context, node, module strin
|
||||
return nil
|
||||
}
|
||||
|
||||
// MaxPayload is the largest message the connected server takes, as the server told this connection.
|
||||
func (b OverNATS) MaxPayload() int64 {
|
||||
if b.Conn == nil {
|
||||
return 0
|
||||
}
|
||||
return b.Conn.MaxPayload()
|
||||
}
|
||||
|
||||
func (b OverNATS) PublishDeclaration(ctx context.Context, node string, body []byte) error {
|
||||
_, err := b.JS.Publish(DeclareSubject(node), body, nats.Context(ctx))
|
||||
if err != nil {
|
||||
|
||||
@@ -24,3 +24,6 @@ func ConnectNats(js *broker.JetStream, enroller Enroller, listener Listener) *Se
|
||||
// Bus is the controller's outbound, whichever transport it connected over. Callers that send a
|
||||
// declaration or ask a tool use this rather than the channel, which one transport does not have.
|
||||
func (s *Server) Bus() Bus { return s.bus }
|
||||
|
||||
// SendingOn is a server that only sends, over the bus given: for a test of a sender with a bus it holds in its hand.
|
||||
func SendingOn(bus Bus) *Server { return &Server{bus: bus} }
|
||||
|
||||
@@ -41,13 +41,54 @@ func Declare(ctx context.Context, bus Bus, signer Signer, node string,
|
||||
return fmt.Errorf("cannot sign a declaration for %s: %w", node, err)
|
||||
}
|
||||
|
||||
body, err := json.Marshal(Signed{Declaration: declaration, Signature: signature})
|
||||
body, err := Envelope(declaration, signature)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// **Measured before it is sent** (novox/hq issue 1689): the bare bus error said only "maximum payload
|
||||
// exceeded", on 2026-10-21 for the control node's declaration, 2 KB over. The limit is the connection's own.
|
||||
if limit := payloadLimitOf(bus); limit > 0 {
|
||||
size := int64(len(body))
|
||||
if DeclarationMeasured != nil {
|
||||
DeclarationMeasured(ctx, node, size, limit)
|
||||
}
|
||||
if size > limit {
|
||||
return fmt.Errorf("%s's declaration is %d bytes on the wire (%d bytes of declaration, signed and encoded), "+
|
||||
"over the bus's largest message of %d bytes by %d: it cannot be sent (novox/hq issue 1689)",
|
||||
node, size, len(declaration), limit, size-limit)
|
||||
}
|
||||
}
|
||||
|
||||
publish, cancel := context.WithTimeout(ctx, timeout)
|
||||
defer cancel()
|
||||
|
||||
return bus.PublishDeclaration(publish, node, body)
|
||||
}
|
||||
|
||||
// Envelope is the message a declaration goes on the bus as: the declaration and its signature, each as base64 in one JSON
|
||||
// document. Its length is what the bus's largest message bounds.
|
||||
func Envelope(declaration, signature []byte) ([]byte, error) {
|
||||
return json.Marshal(Signed{Declaration: declaration, Signature: signature})
|
||||
}
|
||||
|
||||
// NearTheLimit is the share of the bus's largest message at which a declaration is said to be near it: three quarters, so that
|
||||
// a declaration which should stay far from the limit raises the condition while there is still room to act (novox/hq ADR 0486
|
||||
// §8; the first fix of issue 1689 warned at nine tenths while every file was still inline).
|
||||
const NearTheLimit = 0.75
|
||||
|
||||
// DeclarationMeasured is told, at every send of a declaration, its size on the wire and the bus's largest message,
|
||||
// so the controller raises a condition when it nears the limit and clears it when it is back under (issue 1689).
|
||||
// Set by the controller; nil in a test or a tool.
|
||||
var DeclarationMeasured func(ctx context.Context, node string, size, limit int64)
|
||||
|
||||
// payloadLimiter is a bus that knows its largest message.
|
||||
type payloadLimiter interface{ MaxPayload() int64 }
|
||||
|
||||
// payloadLimitOf is the bus's largest message in bytes, or 0 when the bus does not say.
|
||||
func payloadLimitOf(bus Bus) int64 {
|
||||
if l, ok := bus.(payloadLimiter); ok {
|
||||
return l.MaxPayload()
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
//go:build integration
|
||||
|
||||
package link
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/novox/mesh-controller/internal/testbus"
|
||||
)
|
||||
|
||||
// Against a real bus whose largest message is small: the limit is read from the connection, the one that fits goes
|
||||
// through, the one over is refused by us with its size and the limit, not by the client's bare error (issue 1689).
|
||||
func TestADeclarationIsMeasuredAgainstTheRealBusLimit(t *testing.T) {
|
||||
s := testbus.StartWithMaxPayload(t, 2048)
|
||||
conn, err := nats.Connect(s.ClientURL())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer conn.Close()
|
||||
js, err := conn.JetStream()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := js.AddStream(&nats.StreamConfig{Name: "NODES", Subjects: []string{"mesh.node.*.declare"}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
bus := OverNATS{Conn: conn, JS: js}
|
||||
if bus.MaxPayload() != 2048 {
|
||||
t.Fatalf("the limit read from the connection is %d, expected 2048", bus.MaxPayload())
|
||||
}
|
||||
var told []int64
|
||||
was := DeclarationMeasured
|
||||
DeclarationMeasured = func(_ context.Context, _ string, size, limit int64) { told = append(told, size, limit) }
|
||||
t.Cleanup(func() { DeclarationMeasured = was })
|
||||
|
||||
fits := []byte(`{"a":"` + strings.Repeat("x", 600) + `"}`)
|
||||
if err := Declare(t.Context(), bus, fixedSigner{}, "novox", fits, 5*time.Second); err != nil {
|
||||
t.Fatalf("a declaration under the limit: %v", err)
|
||||
}
|
||||
over := []byte(`{"a":"` + strings.Repeat("x", 1600) + `"}`)
|
||||
err = Declare(t.Context(), bus, fixedSigner{}, "novox", over, 5*time.Second)
|
||||
if err == nil || !strings.Contains(err.Error(), "2048 bytes") || strings.Contains(err.Error(), "maximum payload exceeded") {
|
||||
t.Fatalf("a declaration over the limit: %v", err)
|
||||
}
|
||||
if len(told) != 4 || told[1] != 2048 || told[2] <= 2048 {
|
||||
t.Fatalf("measured %v", told)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,57 @@
|
||||
package link
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
type roomBus struct {
|
||||
Bus
|
||||
limit int64
|
||||
sent [][]byte
|
||||
}
|
||||
|
||||
func (b *roomBus) MaxPayload() int64 { return b.limit }
|
||||
func (b *roomBus) PublishDeclaration(_ context.Context, _ string, body []byte) error {
|
||||
b.sent = append(b.sent, body)
|
||||
return nil
|
||||
}
|
||||
|
||||
type fixedSigner struct{}
|
||||
|
||||
func (fixedSigner) Sign(context.Context, []byte) ([]byte, error) { return []byte("sig"), nil }
|
||||
|
||||
// A declaration is measured as it goes on the wire against the bus's own limit; one over it is refused saying size and
|
||||
// limit, and what the measure sees is told to whoever raises a condition (issue 1689).
|
||||
func TestADeclarationOverTheBusLimitIsRefusedSayingSizeAndLimit(t *testing.T) {
|
||||
var seen []int64
|
||||
was := DeclarationMeasured
|
||||
DeclarationMeasured = func(_ context.Context, node string, size, limit int64) {
|
||||
if node != "novox" || limit != 400 {
|
||||
t.Errorf("measured %s against %d", node, limit)
|
||||
}
|
||||
seen = append(seen, size)
|
||||
}
|
||||
t.Cleanup(func() { DeclarationMeasured = was })
|
||||
|
||||
bus := &roomBus{limit: 400}
|
||||
small := []byte(`{"a":"` + strings.Repeat("x", 100) + `"}`)
|
||||
if err := Declare(t.Context(), bus, fixedSigner{}, "novox", small, time.Second); err != nil || len(bus.sent) != 1 {
|
||||
t.Fatalf("a small declaration: %v, %d sent", err, len(bus.sent))
|
||||
}
|
||||
if int64(len(bus.sent[0])) != seen[0] {
|
||||
t.Fatalf("measured %d bytes, sent %d", seen[0], len(bus.sent[0]))
|
||||
}
|
||||
|
||||
big := []byte(`{"a":"` + strings.Repeat("x", 300) + `"}`)
|
||||
err := Declare(t.Context(), bus, fixedSigner{}, "novox", big, time.Second)
|
||||
if err == nil || !strings.Contains(err.Error(), "400 bytes") || !strings.Contains(err.Error(), "novox") ||
|
||||
!strings.Contains(err.Error(), "over") || strings.Contains(err.Error(), "maximum payload exceeded") {
|
||||
t.Fatalf("an oversize declaration: %v", err)
|
||||
}
|
||||
if len(bus.sent) != 1 || len(seen) != 2 || seen[1] <= 400 {
|
||||
t.Fatalf("it was sent (%d) or not measured (%v)", len(bus.sent), seen)
|
||||
}
|
||||
}
|
||||
@@ -56,10 +56,14 @@ var buses sync.Map
|
||||
|
||||
// Start is a server with JetStream, for this test alone: on loopback, on a port the system chooses, its
|
||||
// store in the test's own directory.
|
||||
func Start(t testing.TB) *server.Server {
|
||||
func Start(t testing.TB) *server.Server { return StartWithMaxPayload(t, 0) }
|
||||
|
||||
// StartWithMaxPayload is Start with the largest message the server takes set to maxPayload bytes; zero keeps the
|
||||
// server's default (1 MiB, as the mesh's bus has).
|
||||
func StartWithMaxPayload(t testing.TB, maxPayload int32) *server.Server {
|
||||
t.Helper()
|
||||
opts := &server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(),
|
||||
NoLog: true, NoSigs: true}
|
||||
NoLog: true, NoSigs: true, MaxPayload: maxPayload}
|
||||
s, err := server.NewServer(opts)
|
||||
if err != nil {
|
||||
t.Fatalf("a bus for this test could not be made: %v", err)
|
||||
|
||||
Vendored
+1
-1
@@ -9,7 +9,7 @@ a second place to clean besides the catalogue itself.
|
||||
|
||||
mesh-catalog e0bc919d4945bb5f13f971f74643a0b714cf45da modules/*/module.json, modules/nats/Dockerfile
|
||||
mesh-host 9d9e3a46c4f5943cfffb413d962ce48b75e2fd39 examples/foundation-first-node.lock
|
||||
mesh-sdk 465beb4ea5e37eb36f6ba7fc60274de2d5a0d060 conformance/events/module-event.json
|
||||
mesh-sdk 86f9a1a5c5088b60f694324e84dc31f682156a1b conformance/events/module-event.json
|
||||
|
||||
To move them, from this repository's root, with the two repositories checked out beside it:
|
||||
|
||||
|
||||
Vendored
+1
@@ -167,3 +167,4 @@
|
||||
./cmd/mesh-controller TestAMergeOnTheApproveIntoANewRepositoryAsksTheFirstBuildAndItsOutcomeRegistersTheModule
|
||||
./cmd/mesh-controller TestTheApprovedFirstBuildPassesTheGuardOfAVerbAndIsQueued
|
||||
./cmd/mesh-controller TestAPlanGoesOnWithoutAMachineAwayWhenItsTierIsSent
|
||||
./internal/inventory TestARefusedRegistrationIsKeptByItsBuild
|
||||
|
||||
@@ -2214,6 +2214,7 @@ func parseManifest(raw []byte, predefined PredefinedList) (Manifest, error) {
|
||||
problems = append(problems, replacesProblems(m)...)
|
||||
problems = append(problems, UnitSettingProblems(m)...)
|
||||
problems = append(problems, OwnerSettingProblems(m)...)
|
||||
problems = append(problems, GroupSettingProblems(m)...)
|
||||
problems = append(problems, endpointNameProblems(m)...)
|
||||
problems = append(problems, RouteProblems(m)...)
|
||||
for _, port := range m.Guards {
|
||||
|
||||
+18
-1
@@ -904,7 +904,7 @@ func isModule(catalogue map[string]Manifest, want string) bool {
|
||||
// machine can host the module as far as its capabilities go.
|
||||
func missingCapabilities(m Manifest, node Node) []string {
|
||||
var missing []string
|
||||
for _, c := range m.Capabilities {
|
||||
for _, c := range append(append([]string(nil), m.Capabilities...), impliedCapabilities(m)...) {
|
||||
if !node.Capabilities[c] {
|
||||
missing = append(missing, c)
|
||||
}
|
||||
@@ -1457,3 +1457,20 @@ func packageNames(m map[string][]string) []string {
|
||||
sort.Strings(names)
|
||||
return names
|
||||
}
|
||||
|
||||
// ContainerGroups is the capability a node-engine reports when it reads a container's `groups` and passes each as a
|
||||
// supplementary group (novox/hq ADR 0480 clause 7, issue 1638). An older engine refuses the key (strict decoding).
|
||||
const ContainerGroups = "container-groups"
|
||||
|
||||
// impliedCapabilities are the capabilities a module needs for what its resources say, beyond those it declares: a
|
||||
// container that names groups needs an engine that reads them, so the module is left out of a machine whose report
|
||||
// lacks the capability, naming it, and the field is sent to no engine that would refuse it. None for a module whose
|
||||
// containers name no groups.
|
||||
func impliedCapabilities(m Manifest) []string {
|
||||
for _, r := range m.Resources {
|
||||
if fmt.Sprint(r["type"]) == "container" && len(groupsOf(r)) > 0 {
|
||||
return []string{ContainerGroups}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -66,7 +66,8 @@ var capabilityName = regexp.MustCompile(`^[a-z0-9][a-z0-9-]*$`)
|
||||
// names an `if-capability` may give (novox/hq ADR 0255). A name nothing detects would leave its
|
||||
// contribution out on every machine, silently, so it is refused. A detector added there is added here.
|
||||
var KnownCapabilities = []string{
|
||||
"battery", "container-runtime", "firewall", "graphical-session", "overlay", "package-manager",
|
||||
"battery", "container-groups", "container-runtime", "file-references", "firewall", "graphical-session", "overlay",
|
||||
"package-manager",
|
||||
"power-meter", "privileged", "seat", "service-manager", "uplink-dhcpcd", "uplink-networkmanager",
|
||||
"uplink-systemd-networkd", "user-namespaces", "virtualisation",
|
||||
}
|
||||
|
||||
@@ -85,6 +85,9 @@ func settingInto(resource map[string]any, layers []Layer, module string) error {
|
||||
if fmt.Sprint(resource["type"]) == "directory" {
|
||||
return settingIntoOwner(resource, layers, module)
|
||||
}
|
||||
if fmt.Sprint(resource["type"]) == "container" {
|
||||
return settingIntoGroups(resource, layers, module)
|
||||
}
|
||||
if fmt.Sprint(resource["type"]) != "file" {
|
||||
return nil
|
||||
}
|
||||
@@ -343,6 +346,12 @@ func settingKeysAskedOutsideOwners(m Manifest) map[string]bool {
|
||||
if unit, ok := r["unit"].(string); ok {
|
||||
note(unit)
|
||||
}
|
||||
case "container":
|
||||
for _, g := range groupsOf(r) {
|
||||
if s, ok := g.(string); ok {
|
||||
note(s)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
inValues := func(values map[string]any) {
|
||||
@@ -365,3 +374,100 @@ func settingKeysAskedOutsideOwners(m Manifest) map[string]bool {
|
||||
}
|
||||
return used
|
||||
}
|
||||
|
||||
// A container's supplementary groups may be an operator's value (novox/hq ADR 0480 clause 7, ADR 0348 §11, issue 1636):
|
||||
// n8n joins the media library's group, whose number is the machine's, never the definition's (ADR 0112 §1). Each
|
||||
// element of `groups` is a whole placeholder or a whole number; a placeholder is filled with the number the
|
||||
// config value holds, and the list reaches the node-engine as numbers.
|
||||
|
||||
// groupSide is one element of a container's groups: a whole placeholder or a whole number.
|
||||
var groupSide = regexp.MustCompile(`^(?:\$\{setting:[a-z0-9][a-z0-9_.-]*\}|[0-9]+)$`)
|
||||
|
||||
// groupsOf is a container resource's groups as the manifest wrote them (text or numbers), none when it names none.
|
||||
func groupsOf(r map[string]any) []any {
|
||||
groups, _ := r["groups"].([]any)
|
||||
return groups
|
||||
}
|
||||
|
||||
// GroupSettingProblems refuses, where a manifest is read, a container whose groups hold anything but whole
|
||||
// placeholders and whole numbers, or a placeholder whose key the module neither declares nor asks for elsewhere.
|
||||
func GroupSettingProblems(m Manifest) []string {
|
||||
var problems []string
|
||||
for _, r := range m.Resources {
|
||||
if fmt.Sprint(r["type"]) != "container" {
|
||||
continue
|
||||
}
|
||||
for _, g := range groupsOf(r) {
|
||||
s, isText := g.(string)
|
||||
if !isText {
|
||||
continue // a number: the declaration judges its value
|
||||
}
|
||||
if !groupSide.MatchString(s) {
|
||||
problems = append(problems, fmt.Sprintf("%s: the container %v's group %q is not a whole "+
|
||||
"${config:key} or a whole number: a supplementary group is a number, and nothing else is "+
|
||||
"written into it (novox/hq ADR 0480 clause 7)", m.Module, r["id"], s))
|
||||
continue
|
||||
}
|
||||
if m.Templated() {
|
||||
continue
|
||||
}
|
||||
for _, key := range settingsUsed(s) {
|
||||
if _, declared := m.Settings[key]; !declared && !m.DeclaresConfig() {
|
||||
continue // the settings form judges its own keys
|
||||
}
|
||||
if _, declared := m.Settings[key]; !declared {
|
||||
problems = append(problems, fmt.Sprintf("%s: the container %v's group says ${config:%s}, and the module "+
|
||||
"declares no such value (novox/hq ADR 0348 §4)", m.Module, r["id"], key))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return problems
|
||||
}
|
||||
|
||||
// settingIntoGroups fills ${setting:…} in a container's groups, refusing — with the key and why, the resource left as
|
||||
// it was — a key nothing sets and a value that is not a whole number above zero. A number is kept; the list that
|
||||
// results is numbers alone.
|
||||
func settingIntoGroups(resource map[string]any, layers []Layer, module string) error {
|
||||
groups := groupsOf(resource)
|
||||
if len(groups) == 0 {
|
||||
return nil
|
||||
}
|
||||
filled := make([]any, 0, len(groups))
|
||||
for _, g := range groups {
|
||||
s, isText := g.(string)
|
||||
if !isText {
|
||||
if f, isNumber := g.(float64); isNumber && f == float64(int(f)) {
|
||||
g = int(f) // a number the manifest wrote, as the numbers a placeholder gives
|
||||
}
|
||||
filled = append(filled, g)
|
||||
continue
|
||||
}
|
||||
keys := settingsUsed(s)
|
||||
if len(keys) != 1 || settingRef.ReplaceAllString(s, "") != "" {
|
||||
filled = append(filled, g) // not a whole placeholder: the manifest check refuses it
|
||||
continue
|
||||
}
|
||||
key := keys[0]
|
||||
value, set := settingValue(layers, key)
|
||||
if !set {
|
||||
return &UnsetSettingError{Module: module, Setting: key,
|
||||
where: fmt.Sprintf("a container %v whose groups", resource["id"]), setToday: orNoSettings(layers),
|
||||
said: fmt.Sprintf(
|
||||
"%s has a container %v whose groups say ${setting:%s}, and nothing sets %q for it — an operator's "+
|
||||
"value is the assignment's, never the definition's (novox/hq ADR 0112): "+
|
||||
"`settings set %s <file>` with {%q: …}%s",
|
||||
module, resource["id"], key, key, module, key, orNoSettings(layers))}
|
||||
}
|
||||
v := plainly(value)
|
||||
id, whole := wholeID(v)
|
||||
if !whole || id == 0 {
|
||||
return fmt.Errorf("%s: the setting %q is %q, which cannot be a group of the container %v: a supplementary "+
|
||||
"group is a whole number above zero, never a name or other text (novox/hq ADR 0480 clause 7, ADR 0328 §15)",
|
||||
module, key, v, resource["id"])
|
||||
}
|
||||
filled = append(filled, int(id))
|
||||
}
|
||||
resource["groups"] = filled
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -56,7 +56,7 @@ type filler struct {
|
||||
// seatInto, machineInto and contributionsInto. A pass that comes to read another field adds it here,
|
||||
// or the sweep refuses the placeholder it would have filled.
|
||||
var fillers = []filler{
|
||||
{namespace: "setting", patterns: []*regexp.Regexp{settingRef}, fields: map[string][]string{"file": {"content"}, "service": {"unit"}, "directory": {"owner"}}},
|
||||
{namespace: "setting", patterns: []*regexp.Regexp{settingRef}, fields: map[string][]string{"file": {"content"}, "service": {"unit"}, "directory": {"owner"}, "container": {"groups"}}},
|
||||
{namespace: "dir", patterns: []*regexp.Regexp{dirRef}, fields: map[string][]string{"*": {"path", "content", "volumes", "env", "env-file"}}},
|
||||
{namespace: "access", patterns: []*regexp.Regexp{accessRef}, fields: map[string][]string{"*": {"path", "content", "volumes", "env", "env-file"}}},
|
||||
{namespace: "secret", patterns: []*regexp.Regexp{placeholder}, fields: map[string][]string{"file": {"content"}},
|
||||
|
||||
@@ -324,6 +324,15 @@ type File struct {
|
||||
// depend on a field somewhere else.
|
||||
Bytes string `json:"bytes,omitempty"`
|
||||
|
||||
// Digest and Size say the content is not carried but fetched: a file over FileReferenceThreshold travels as
|
||||
// the sha256 digest of its content, "sha256:<64 hex>", and its size in bytes, in place of Content or Bytes
|
||||
// (novox/hq ADR 0486 §1). The node-engine fetches the object of that name from FileBucketFor(node), checks what it
|
||||
// received against Digest and Size, and only then writes the file. Exclusive with Content, Sealed, Bytes,
|
||||
// VaultSealed and Secrets; sent only to an engine that reports FileReferencesCapability, as an older engine
|
||||
// refuses a field it does not know, whole.
|
||||
Digest string `json:"digest,omitempty"`
|
||||
Size int64 `json:"size,omitempty"`
|
||||
|
||||
// Owner is the user this belongs to, by name. Absent means root.
|
||||
Owner string `json:"owner,omitempty"`
|
||||
|
||||
@@ -465,6 +474,10 @@ func (f *File) validate(where string, _ bool) []string {
|
||||
problems = append(problems, where+": a file the vault sealed whole has no holes for secrets")
|
||||
}
|
||||
}
|
||||
problems = append(problems, f.checkReference(where)...)
|
||||
if f.Digest != "" || f.Size != 0 {
|
||||
said = append(said, "digest")
|
||||
}
|
||||
if len(said) > 1 {
|
||||
sort.Strings(said)
|
||||
problems = append(problems, where+
|
||||
@@ -1454,6 +1467,14 @@ type Container struct {
|
||||
// container runs as its image says. Its key is `identity`, the manifest's word; the field is not, because
|
||||
// a resource's Identity is its id.
|
||||
RunAs *ContainerIdentity `json:"identity,omitempty"`
|
||||
|
||||
// Groups are the supplementary groups the container's process joins, as numbers, each passed to the runtime
|
||||
// as a group to add (novox/hq ADR 0480 clause 7, issue 1636): n8n writes the media library through the
|
||||
// library's group, which a stock image knows nothing of. Each is above zero and below the kernel's "no group"
|
||||
// (ADR 0328 §15: never root). Part of the spec, so a change recreates the container. Absent: none are added.
|
||||
// An older node-engine refuses the key (strict decoding), so the controller sends it only to a node that
|
||||
// reports it reads it.
|
||||
Groups []int64 `json:"groups,omitempty"`
|
||||
}
|
||||
|
||||
// The ways an image is told who to run as (novox/hq ADR 0325 §2).
|
||||
@@ -1521,6 +1542,30 @@ func (i *ContainerIdentity) problems(where string, env map[string]string) []stri
|
||||
return problems
|
||||
}
|
||||
|
||||
// groupProblems refuses a supplementary group that is root's or below it, the kernel's "no group" number, or a
|
||||
// number named twice (novox/hq ADR 0480 clause 7, ADR 0328 §15).
|
||||
func (c *Container) groupProblems(where string) []string {
|
||||
var problems []string
|
||||
seen := map[int64]bool{}
|
||||
for _, g := range c.Groups {
|
||||
switch {
|
||||
case g == 0:
|
||||
problems = append(problems, fmt.Sprintf("%s: container %s adds group 0, and the mesh never makes or keeps anything "+
|
||||
"as root: a supplementary group is never root's (novox/hq ADR 0328 §15)", where, c.Name))
|
||||
case g < 0:
|
||||
problems = append(problems, fmt.Sprintf("%s: container %s adds group %d, below zero: a group is a whole number "+
|
||||
"above zero", where, c.Name, g))
|
||||
case g >= 1<<32-1:
|
||||
problems = append(problems, fmt.Sprintf("%s: container %s adds group %d, which is the kernel's \"no group\" "+
|
||||
"number or beyond it", where, c.Name, g))
|
||||
case seen[g]:
|
||||
problems = append(problems, fmt.Sprintf("%s: container %s adds group %d twice", where, c.Name, g))
|
||||
}
|
||||
seen[g] = true
|
||||
}
|
||||
return problems
|
||||
}
|
||||
|
||||
func (c *Container) Identity() string { return c.ID }
|
||||
func (c *Container) Kind() Type { return TypeContainer }
|
||||
func (c *Container) Target() string { return c.Name }
|
||||
@@ -1603,6 +1648,7 @@ func (c *Container) validate(where string, _ bool) []string {
|
||||
"runs and exits, and joins nothing afterwards")
|
||||
}
|
||||
problems = append(problems, c.RunAs.problems(where+": container "+c.Name, c.Env)...)
|
||||
problems = append(problems, c.groupProblems(where)...)
|
||||
return append(problems, checkImage(where, c.Image)...)
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,150 @@
|
||||
package declaration
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"encoding/base64"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"regexp"
|
||||
)
|
||||
|
||||
// The three names of a file travelling by digest (novox/hq ADR 0486 §1, §2, §5), shared by the node-engine that
|
||||
// reads them and the controller that writes them.
|
||||
const (
|
||||
// FileReferenceThreshold is the size above which a file travels by digest: a file of this many bytes or fewer
|
||||
// stays in the declaration as it is.
|
||||
FileReferenceThreshold = 64 * 1024
|
||||
// FileBucketPrefix starts the name of a node's object bucket, FileBucketFor(node), which holds the contents of the
|
||||
// files of that node's declaration, each under the hex of its sha256 digest (ObjectName). One bucket per node,
|
||||
// so a grant names it and a literal `*` is never needed (ADR 0486 §2, changed by ADR 0488 §1 and §2). The controller alone writes it; the
|
||||
// node's engine reads an object by name.
|
||||
FileBucketPrefix = "declaration-files-"
|
||||
// FileStreamPrefix starts the name of the stream behind such a bucket, which the bus's grants name.
|
||||
FileStreamPrefix = "OBJ_"
|
||||
// FileReferencesCapability is what a node-engine reports when it reads a file given as a digest and size.
|
||||
FileReferencesCapability = "file-references"
|
||||
)
|
||||
|
||||
var nodeToken = regexp.MustCompile(`^[A-Za-z0-9_-]+$`)
|
||||
|
||||
// FileBucketFor is the name of the object bucket of a node's files, "declaration-files-<node>" (novox/hq ADR 0488
|
||||
// §4, which replaces the single bucket constant of issue 1694). The node name must be a name the bus accepts in a
|
||||
// bucket and in a subject token: letters, digits, '_' and '-', so no '.', '*', '>' or space.
|
||||
func FileBucketFor(node string) (string, error) {
|
||||
if !nodeToken.MatchString(node) {
|
||||
return "", fmt.Errorf("node %q; a node's bucket is named by letters, digits, '_' and '-' only", node)
|
||||
}
|
||||
return FileBucketPrefix + node, nil
|
||||
}
|
||||
|
||||
// FileStreamFor is the name of the stream behind a node's bucket, "OBJ_" and the bucket's name.
|
||||
func FileStreamFor(node string) (string, error) {
|
||||
b, err := FileBucketFor(node)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return FileStreamPrefix + b, nil
|
||||
}
|
||||
|
||||
var digestForm = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`)
|
||||
|
||||
// DigestOf is the digest a reference names for a content: "sha256:<64 lower-case hex>".
|
||||
func DigestOf(content []byte) string {
|
||||
sum := sha256.Sum256(content)
|
||||
return "sha256:" + hex.EncodeToString(sum[:])
|
||||
}
|
||||
|
||||
// ObjectName is the name of the object holding the content a digest names: the hex without its "sha256:" prefix.
|
||||
func ObjectName(digest string) (string, error) {
|
||||
if !digestForm.MatchString(digest) {
|
||||
return "", fmt.Errorf("digest %q is not sha256:<64 lower-case hex characters>", digest)
|
||||
}
|
||||
return digest[len("sha256:"):], nil
|
||||
}
|
||||
|
||||
// IsReference reports whether the file gives its content as a digest and size instead of carrying it.
|
||||
func (f *File) IsReference() bool { return f.Digest != "" }
|
||||
|
||||
// checkReference judges the digest and size, and what a reference may not be given with.
|
||||
func (f *File) checkReference(where string) []string {
|
||||
if f.Digest == "" && f.Size == 0 {
|
||||
return nil
|
||||
}
|
||||
var problems []string
|
||||
if !digestForm.MatchString(f.Digest) {
|
||||
problems = append(problems, fmt.Sprintf("%s: digest %q; a file by reference names its content as "+
|
||||
"sha256:<64 lower-case hex characters>", where, f.Digest))
|
||||
}
|
||||
if f.Size <= 0 {
|
||||
problems = append(problems, fmt.Sprintf("%s: size %d; a file by reference states its size in bytes, "+
|
||||
"more than zero", where, f.Size))
|
||||
}
|
||||
if f.Content != "" || f.Bytes != "" || f.Sealed != "" || f.VaultSealed != nil || len(f.Secrets) > 0 {
|
||||
problems = append(problems, where+": a file by reference carries its content in the object its digest "+
|
||||
"names, so it says no content, bytes, sealed, vault-sealed or secrets")
|
||||
}
|
||||
if f.Into != "" {
|
||||
problems = append(problems, where+": a file by reference is written whole; a file written into says its "+
|
||||
"part in content")
|
||||
}
|
||||
return problems
|
||||
}
|
||||
|
||||
// Body is the bytes the file's content stands for: its content, or its bytes decoded.
|
||||
func (f *File) Body() ([]byte, error) {
|
||||
if f.Bytes == "" {
|
||||
return []byte(f.Content), nil
|
||||
}
|
||||
b, err := base64.StdEncoding.DecodeString(f.Bytes)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("file %s: bytes are not base64: %w", f.ID, err)
|
||||
}
|
||||
return b, nil
|
||||
}
|
||||
|
||||
// TravelsByDigest reports whether the controller may put this file's content in the bucket: it is plain content
|
||||
// or bytes (not sealed, not filled with secrets, not written into another file) of more than FileReferenceThreshold
|
||||
// bytes.
|
||||
func (f *File) TravelsByDigest() bool {
|
||||
if f.IsReference() || f.Sealed != "" || f.VaultSealed != nil || len(f.Secrets) > 0 || f.Into != "" {
|
||||
return false
|
||||
}
|
||||
if f.Bytes != "" {
|
||||
// DecodedLen counts padding, so the exact length is taken off the decoded bytes.
|
||||
body, err := f.Body()
|
||||
return err == nil && len(body) > FileReferenceThreshold
|
||||
}
|
||||
return len(f.Content) > FileReferenceThreshold
|
||||
}
|
||||
|
||||
// AsReference is the file as it is sent to an engine that reports FileReferencesCapability: a copy with the digest
|
||||
// and size in place of the content, and the content the controller puts into the bucket under ObjectName(digest).
|
||||
// A file that does not travel by digest is refused, not passed through, so a caller cannot send a small file empty.
|
||||
func (f *File) AsReference() (File, []byte, error) {
|
||||
if !f.TravelsByDigest() {
|
||||
return File{}, nil, fmt.Errorf("file %s does not travel by digest", f.ID)
|
||||
}
|
||||
body, err := f.Body()
|
||||
if err != nil {
|
||||
return File{}, nil, err
|
||||
}
|
||||
ref := *f
|
||||
ref.Content, ref.Bytes = "", ""
|
||||
ref.Digest, ref.Size = DigestOf(body), int64(len(body))
|
||||
return ref, body, nil
|
||||
}
|
||||
|
||||
// Verify checks what was fetched against the digest and size the file names; the node-engine writes the file only
|
||||
// after it returns nil. A mismatch is the apply's failure, never a partial apply (ADR 0486 §1).
|
||||
func (f *File) Verify(body []byte) error {
|
||||
if !f.IsReference() {
|
||||
return fmt.Errorf("file %s is not given by reference", f.ID)
|
||||
}
|
||||
if int64(len(body)) != f.Size {
|
||||
return fmt.Errorf("file %s: the object is %d bytes and the declaration says %d", f.ID, len(body), f.Size)
|
||||
}
|
||||
if got := DigestOf(body); got != f.Digest {
|
||||
return fmt.Errorf("file %s: the object hashes to %s and the declaration names %s", f.ID, got, f.Digest)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -25,7 +25,7 @@ type BuildRecord struct {
|
||||
}
|
||||
|
||||
// Artifact is one thing a build made: mesh-controller internal/chain/records.go Artifact, json keys name, kind and
|
||||
// reference, nothing else (novox/hq ADR 0478 clause 7).
|
||||
// reference, nothing else (novox/hq ADR 0478 clause 8).
|
||||
type Artifact struct {
|
||||
Name string `json:"name"`
|
||||
Kind string `json:"kind"`
|
||||
|
||||
+16
-2
@@ -389,11 +389,20 @@ type Module struct {
|
||||
Src uint64 `json:"src,omitempty"`
|
||||
Repository string `json:"repository"`
|
||||
Commit string `json:"commit"`
|
||||
// Path is the module's directory in its repository, trimmed of slashes, empty when its manifest is the repository's
|
||||
// root (novox/hq ADR 0478 clause 8): the build tools need it. A value written before has none.
|
||||
// Path is the module's directory in its repository, trimmed of slashes, and RootPath (".") when its manifest is the
|
||||
// repository's root (novox/hq ADR 0478 clause 9, issue 1615): the build tools need it. Empty means not known: a value
|
||||
// written before the indexer learned paths has none.
|
||||
Path string `json:"path,omitempty"`
|
||||
// Toolchains are the toolchains FP2 was computed with, by language (go, typescript): the version installed and the
|
||||
// build module's recipe version the build nodes agreed on then (novox/hq ADR 0477 clause 2, ADR 0478 clause 3). A builder
|
||||
// builds the fingerprint only with these. Empty when FP2 is empty or the module has no toolchain; a value written
|
||||
// before has none (issue 1615).
|
||||
Toolchains map[string]ToolchainUsed `json:"toolchains,omitempty"`
|
||||
}
|
||||
|
||||
// RootPath is Module.Path of a module whose manifest is its repository's root.
|
||||
const RootPath = "."
|
||||
|
||||
// KeptInputs2 are SourceInputs2 without the module's name (the key has it) and without toolchains (those come from
|
||||
// the build modules' toolchain keys when the fingerprint is computed): what a Module keeps of a merge.
|
||||
type KeptInputs2 struct {
|
||||
@@ -421,6 +430,11 @@ func (m Module) check() error {
|
||||
case m.FP2 != "" && m.None2 != "":
|
||||
return fmt.Errorf("delivery: a module has both fp2 and none2")
|
||||
}
|
||||
for language, t := range m.Toolchains {
|
||||
if t.Version == "" || t.Recipe == "" {
|
||||
return fmt.Errorf("delivery: the module's toolchain of %q needs its version and its recipe version", language)
|
||||
}
|
||||
}
|
||||
if m.Inputs2 != nil && m.Manifest == "" {
|
||||
return fmt.Errorf("delivery: a module keeps new-form inputs but has no manifest digest")
|
||||
}
|
||||
|
||||
Vendored
+1
-1
@@ -1,4 +1,4 @@
|
||||
# git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021093645-465beb4ea5e3
|
||||
# git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021134822-86f9a1a5c508
|
||||
## explicit; go 1.26.0
|
||||
git.novox.be/novox/mesh-sdk/go/asks
|
||||
git.novox.be/novox/mesh-sdk/go/catalogue
|
||||
|
||||
Reference in New Issue
Block a user