diff --git a/cmd/mesh-controller/busobjects.go b/cmd/mesh-controller/busobjects.go index f56bdc47..f7213d32 100644 --- a/cmd/mesh-controller/busobjects.go +++ b/cmd/mesh-controller/busobjects.go @@ -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. diff --git a/cmd/mesh-controller/compose_sdk.go b/cmd/mesh-controller/compose_sdk.go index 02aae949..8eb0b66f 100644 --- a/cmd/mesh-controller/compose_sdk.go +++ b/cmd/mesh-controller/compose_sdk.go @@ -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) diff --git a/cmd/mesh-controller/declaration_capability_integration_test.go b/cmd/mesh-controller/declaration_capability_integration_test.go new file mode 100644 index 00000000..f408079c --- /dev/null +++ b/cmd/mesh-controller/declaration_capability_integration_test.go @@ -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) + } + } + } +} diff --git a/cmd/mesh-controller/declaration_files.go b/cmd/mesh-controller/declaration_files.go new file mode 100644 index 00000000..c87902ab --- /dev/null +++ b/cmd/mesh-controller/declaration_files.go @@ -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) + } + } +} diff --git a/cmd/mesh-controller/declaration_files_integration_test.go b/cmd/mesh-controller/declaration_files_integration_test.go new file mode 100644 index 00000000..b554e115 --- /dev/null +++ b/cmd/mesh-controller/declaration_files_integration_test.go @@ -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") + } +} diff --git a/cmd/mesh-controller/declaration_files_test.go b/cmd/mesh-controller/declaration_files_test.go new file mode 100644 index 00000000..c428d418 --- /dev/null +++ b/cmd/mesh-controller/declaration_files_test.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) + } +} diff --git a/cmd/mesh-controller/declaration_room.go b/cmd/mesh-controller/declaration_room.go index 827348c4..2eb0b6df 100644 --- a/cmd/mesh-controller/declaration_room.go +++ b/cmd/mesh-controller/declaration_room.go @@ -13,8 +13,8 @@ import ( // **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 90% of the limit, urgent over it, cleared when -// the declaration is back under 90%. The limit is the connection's own, not a constant. +// 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" diff --git a/cmd/mesh-controller/declaration_room_test.go b/cmd/mesh-controller/declaration_room_test.go index c1aba998..a52b547f 100644 --- a/cmd/mesh-controller/declaration_room_test.go +++ b/cmd/mesh-controller/declaration_room_test.go @@ -33,7 +33,7 @@ func TestTheVaultsExportIsSentCompactAndNothingElseIsTouched(t *testing.T) { } } -// From 90% of the bus's largest message the declaration is a warning; over it, urgent; back under 90%, cleared. +// 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, @@ -50,22 +50,22 @@ func TestADeclarationNearTheBusLimitRaisesAConditionAndClearsWhenSmaller(t *test return c, found } - settleDeclarationRoom(t.Context(), k, "novox", 899, limit) + settleDeclarationRoom(t.Context(), k, "novox", 749, limit) if _, found := read(); found { - t.Fatal("a declaration at 89.9% raised a condition") + t.Fatal("a declaration at 74.9% raised a condition") } - settleDeclarationRoom(t.Context(), k, "novox", 900, limit) + settleDeclarationRoom(t.Context(), k, "novox", 750, limit) c, found := read() - if !found || c.Severity != conditions.Warning || !strings.Contains(c.Summary, "900 bytes") || + if !found || c.Severity != conditions.Warning || !strings.Contains(c.Summary, "750 bytes") || !strings.Contains(c.Summary, "1000") || c.Subject.Machine != "novox" { - t.Fatalf("at 90%% the condition is %v %+v", found, c) + 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", 850, limit) + settleDeclarationRoom(t.Context(), k, "novox", 700, limit) if _, found := read(); found { - t.Fatal("the condition stayed after the declaration fell under 90%") + t.Fatal("the condition stayed after the declaration fell under 75%") } } diff --git a/cmd/mesh-controller/declaration_sweep.go b/cmd/mesh-controller/declaration_sweep.go new file mode 100644 index 00000000..33bc2a79 --- /dev/null +++ b/cmd/mesh-controller/declaration_sweep.go @@ -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()) +} diff --git a/cmd/mesh-controller/plain_words.go b/cmd/mesh-controller/plain_words.go index 0c69739b..41ebf533 100644 --- a/cmd/mesh-controller/plain_words.go +++ b/cmd/mesh-controller/plain_words.go @@ -195,6 +195,14 @@ 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 { diff --git a/cmd/mesh-controller/push.go b/cmd/mesh-controller/push.go index 08aa18ac..e3c12494 100644 --- a/cmd/mesh-controller/push.go +++ b/cmd/mesh-controller/push.go @@ -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, diff --git a/cmd/mesh-controller/sendable.go b/cmd/mesh-controller/sendable.go index ded06fac..446ebf06 100644 --- a/cmd/mesh-controller/sendable.go +++ b/cmd/mesh-controller/sendable.go @@ -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 +} diff --git a/go.mod b/go.mod index dd814bad..dcb6ff56 100644 --- a/go.mod +++ b/go.mod @@ -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.20261021144048-d4deecdb380b + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021153908-725590e06e75 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 diff --git a/go.sum b/go.sum index e67f5187..f06c6de7 100644 --- a/go.sum +++ b/go.sum @@ -1,7 +1,7 @@ git.novox.be/novox/mesh-host v0.0.0-20261018171808-9d9e3a46c4f5 h1:E5stfIeCN6uYgA9TCv+4K8JgaURg3zQppvUgJLaDMP4= git.novox.be/novox/mesh-host v0.0.0-20261018171808-9d9e3a46c4f5/go.mod h1:4wyoCr6IIIIarEUuROThWDj7DAvJvIidqMjWcpSnOwE= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021144048-d4deecdb380b h1:5UlQ8ytlHNgqlm5AbwJEnZwuf1EWUiNyXX2hcOV9Yro= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021144048-d4deecdb380b/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021153908-725590e06e75 h1:AcYfyJpYaUufhI93VG76+b001msXY4XX7kuRjnf33HY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021153908-725590e06e75/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= diff --git a/internal/broker/declaration_files.go b/internal/broker/declaration_files.go new file mode 100644 index 00000000..096684a2 --- /dev/null +++ b/internal/broker/declaration_files.go @@ -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-`. +// +// **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..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_, whose messages are the objects' metadata +// (`$O..M.`) and chunks (`$O..C.`). + +// 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..$O.B.C.`. **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 +} diff --git a/internal/broker/declaration_files_live_test.go b/internal/broker/declaration_files_live_test.go new file mode 100644 index 00000000..e21f6e7d --- /dev/null +++ b/internal/broker/declaration_files_live_test.go @@ -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 +} diff --git a/internal/broker/declaration_files_test.go b/internal/broker/declaration_files_test.go new file mode 100644 index 00000000..0e584dc3 --- /dev/null +++ b/internal/broker/declaration_files_test.go @@ -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) + } +} diff --git a/internal/broker/nats.go b/internal/broker/nats.go index 71380f2d..c9007a14 100644 --- a/internal/broker/nats.go +++ b/internal/broker/nats.go @@ -104,6 +104,10 @@ type Principal struct { // Username and inbox are ... 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 { diff --git a/internal/broker/testdata/composed.conf b/internal/broker/testdata/composed.conf index f0d25f8d..1e2b427d 100644 --- a/internal/broker/testdata/composed.conf +++ b/internal/broker/testdata/composed.conf @@ -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" } } } diff --git a/internal/broker/users.go b/internal/broker/users.go index c33814db..a63e281e 100644 --- a/internal/broker/users.go +++ b/internal/broker/users.go @@ -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 diff --git a/internal/broker/writers.go b/internal/broker/writers.go index 07984fdd..06a44fde 100644 --- a/internal/broker/writers.go +++ b/internal/broker/writers.go @@ -219,6 +219,13 @@ var WritersTable = []WriterRow{ {State: "a machine's have", Writer: "the node-engine of the machine, under its own keys `.>`", 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 + "- 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}, diff --git a/internal/broker/writers_test.go b/internal/broker/writers_test.go index 500e44b4..fd56ff3f 100644 --- a/internal/broker/writers_test.go +++ b/internal/broker/writers_test.go @@ -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). diff --git a/internal/link/builds_nats.go b/internal/link/builds_nats.go index f95a2295..b901e9fe 100644 --- a/internal/link/builds_nats.go +++ b/internal/link/builds_nats.go @@ -362,6 +362,14 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build)) _ = msg.Nak() continue } + // **A duplicate of a check another builder is running is dropped** (issue 1649): an ask redelivered to this machine + // because its first holder did not say, in time, that it had it. The running key names that builder and this same + // ask; running it a second time would spend a second place on one check. + if why := m.runningElsewhere(ctx, request); why != "" { + fmt.Fprintf(os.Stderr, "%s is handed out again while another builder runs it and is dropped: %s\n", request.ID, why) + _ = msg.Ack() + continue + } // A build outlives the acknowledgement window many times over; said while it runs, // as the controller says it for its own long handlers, so the server neither hands // the ask to a second machine nor counts the wait against its deliveries. @@ -376,6 +384,35 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build)) } } +// runningElsewhere says why an ask need not run here, or nothing: the bucket of the checks running says another builder +// runs this very ask (its id, its head and its trunk commit). A key that cannot be read runs the check. +func (m *natsMachine) runningElsewhere(ctx context.Context, r BuildRequest) string { + if r.Check == nil { + return "" + } + running, ok, err := ReadRunning(ctx, m.js.Conn(), r) + if err != nil { + fmt.Fprintf(os.Stderr, "could not read whether %s is running elsewhere, so it runs: %v\n", r.ID, err) + return "" + } + return duplicateOf(running, ok, r, m.on) +} + +// runningFreshFor is how recent a running key's last heartbeat must be for its builder to be taken as alive: two missed +// heartbeats of ten seconds are not yet a dead builder, and a redelivery after a crash comes a minute after the last one. +const runningFreshFor = 25 * time.Second + +// duplicateOf says why an ask is a second hand-out of one another builder is running, or nothing. +func duplicateOf(running CheckRunning, found bool, r BuildRequest, me string) string { + // Seen is the server's time and runningFreshFor is judged against this builder's clock: a skew over about 35 s would matter. + // A key not renewed for runningFreshFor is a builder that has gone: its ask is redelivered after the ack wait, and + // that is the recovery, not a duplicate (the key itself lives two minutes). + if found && running.Build == r.ID && running.Builder != me && running.Same(r) && time.Since(running.Seen) < runningFreshFor { + return fmt.Sprintf("%s is running it as %s since %s", running.Builder, running.Build, running.Started.Local().Format("15:04:05")) + } + return "" +} + // bind binds a queue's worker, which exists and is not created here. // // **The consumer's own filter, not the one subject this machine cares about.** The client checks what is asked for diff --git a/internal/link/bus_nats.go b/internal/link/bus_nats.go index bfaebdb4..2aebcb1d 100644 --- a/internal/link/bus_nats.go +++ b/internal/link/bus_nats.go @@ -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} } diff --git a/internal/link/checks_nats.go b/internal/link/checks_nats.go index 176c70ec..b720bbf4 100644 --- a/internal/link/checks_nats.go +++ b/internal/link/checks_nats.go @@ -253,11 +253,25 @@ func watchPullIn(ctx context.Context, buckets []string, open func(ctx context.Co // checkTakeWait is how long a taking loop waits on one queue for an ask that is not there, and takeIdleWait how long it // waits on the first when all three are empty. +// +// **A fetch's pull request must end before the fetch does** (issue 1649). nats.go gives the server-side request an expiry a tenth +// shorter than the fetch's wait (fetchExpiryOf), and the server then answers with a status message at the expiry; a wait so +// short that the answer can land after the fetch gave up left a status message in the pull's buffer, and anything a request +// could still deliver would be held by a machine that is busy. Each wait leaves at least minFetchMargin between the two. const ( - checkTakeWait = 30 * time.Millisecond + checkTakeWait = 100 * time.Millisecond takeIdleWait = time.Second ) +// minFetchMargin is the least time between a fetch's server-side expiry and the end of its wait. +const minFetchMargin = 10 * time.Millisecond + +// fetchExpiryOf is the expiry nats.go gives the pull request of a fetch that waits this long: the wait less a tenth of it, or +// less five seconds. +func fetchExpiryOf(wait time.Duration) time.Duration { + return wait - min(wait/10, 5*time.Second) +} + // CheckRunning is what a builder says of the check it is running, in the bucket of the checks running (novox/hq ADR 0406 // §3): the head and trunk commit it checks, the builder and its build. The key is the pull request's (RunningKeyOf). type CheckRunning struct { @@ -268,6 +282,8 @@ type CheckRunning struct { // Started is when the check started, kept on every heartbeat: the entry's own time is the last heartbeat's, and a // reader that asks how long a check has run reads this (novox/hq ADR 0433 §6). Started time.Time `json:"started"` + // Seen is when the key was last put, which is the last heartbeat: read from the entry, never written (issue 1649). + Seen time.Time `json:"-"` } // RunningKeyOf is the key a check is said running under: its pull request's tokens, or `group.` for a group's @@ -319,6 +335,7 @@ func ReadRunning(ctx context.Context, conn *nats.Conn, r BuildRequest) (CheckRun if err := json.Unmarshal(e.Value(), &running); err != nil { return CheckRunning{}, false, nil // a key of a builder from before this means no match } + running.Seen = e.Created() return running, true, nil } diff --git a/internal/link/checks_nats_integration_test.go b/internal/link/checks_nats_integration_test.go index 71959b63..5e535632 100644 --- a/internal/link/checks_nats_integration_test.go +++ b/internal/link/checks_nats_integration_test.go @@ -6,6 +6,7 @@ import ( "context" "encoding/json" "reflect" + "strings" "testing" "time" @@ -510,3 +511,122 @@ func TestAnUnacknowledgedCheckGoesToASecondBuilder(t *testing.T) { stop() } } + +// An ask handed out a second time to another builder while the first still runs it is acknowledged and not run: the +// running key names the other builder and this same ask (issue 1649). +func TestAnAskHandedOutAgainWhileAnotherBuilderRunsItIsNotRunTwice(t *testing.T) { + js := aBusWithTheBuildRole(t) + aKeyBucket(t, js) + aRunningBucket(t, js, 2*time.Minute) + ctx, stop := context.WithCancel(t.Context()) + defer stop() + asks := BuildsOn(js, TheBuildMachine) + running := make(chan string, 4) + release := make(chan struct{}) + a := MachineOverNATS(js, "builder-a") + defer a.Close() + go func() { + _ = a.Take(ctx, func(ctx context.Context, work Build) { + running <- work.Request().ID + <-release + _ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "builder-a", Failed: "no"}) + _ = work.Done() + }) + }() + check := aPullCheck("build-twice", 7, "head-7-000000000", "trunk-1-000000", false) + if err := asks.Ask(ctx, check); err != nil { + t.Fatal(err) + } + select { + case <-running: + case <-time.After(10 * time.Second): + t.Fatal("builder a took nothing") + } + // The bus's redelivery, as it reaches the other machine: the same ask, the same id. + if err := asks.Ask(ctx, check); err != nil { + t.Fatal(err) + } + b := MachineOverNATS(js, "builder-b") + defer b.Close() + took := make(chan string, 4) + go func() { + _ = b.Take(ctx, func(ctx context.Context, work Build) { + took <- work.Request().ID + _ = work.Done() + }) + }() + select { + case id := <-took: + t.Fatalf("builder b ran %s, which builder a is running", id) + case <-time.After(3 * time.Second): + } + if n := len(queueOf(t, js).Of(AskWaiting)); n != 0 { + t.Errorf("%d request(s) still wait; builder b was to acknowledge the duplicate", n) + } + close(release) +} + +// A busy machine fetches nothing, so no ask waiting in the queue is taken by it, whatever status message sits in its pulls +// (issue 1649). A pull request of our own that expires at once leaves such a message in the other checks' pull. +func TestABusyMachineTakesNoWaitingAskWhateverItsPullsHold(t *testing.T) { + js := aBusWithTheBuildRole(t) + aKeyBucket(t, js) + aRunningBucket(t, js, 2*time.Minute) + ctx, stop := context.WithCancel(t.Context()) + defer stop() + asks := BuildsOn(js, TheBuildMachine) + busy := make(chan string, 4) + release := make(chan struct{}) + a := MachineOverNATS(js, "builder-a").(*natsMachine) + defer a.Close() + go func() { + _ = a.Take(ctx, func(ctx context.Context, work Build) { + busy <- work.Request().ID + <-release + _ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "builder-a", Failed: "no"}) + _ = work.Done() + }) + }() + if err := asks.Ask(ctx, BuildRequest{ID: "busy", Repository: "/d"}); err != nil { + t.Fatal(err) + } + select { + case <-busy: + case <-time.After(10 * time.Second): + t.Fatal("builder a took nothing") + } + var other *takeQueue + for _, q := range a.subs { + if q.kind == queueOther && q.bound { + other = q + } + } + if other == nil { + t.Fatal("builder a has no pull on the other checks") + } + // The queue is empty, so this request is answered at its expiry with a status message that lands in the pull. + reply := strings.TrimSuffix(other.sub.Subject, "*") + "late" + if err := js.Conn().PublishRequest("$JS.API.CONSUMER.MSG.NEXT."+other.worker.Stream+"."+other.worker.Name, reply, + []byte(`{"batch":1,"expires":1000000}`)); err != nil { + t.Fatal(err) + } + time.Sleep(500 * time.Millisecond) + for i, id := range []string{"wait-1", "wait-2"} { + if err := asks.Ask(ctx, aPullCheck(id, 20+i, "head-x-000000000", "trunk-1-000000", false)); err != nil { + t.Fatal(err) + } + } + time.Sleep(5 * time.Second) + info, err := js.Context().ConsumerInfo(other.worker.Stream, other.worker.Name) + if err != nil { + t.Fatal(err) + } + if info.NumAckPending != 0 || info.NumRedelivered != 0 || info.Delivered.Consumer != 0 { + t.Fatalf("a busy machine took from the queue: %d delivered, %d pending, %d redelivered", info.Delivered.Consumer, + info.NumAckPending, info.NumRedelivered) + } + if n := len(queueOf(t, js).Of(AskWaiting)); n != 2 { + t.Fatalf("%d asks wait; want the 2 nobody took", n) + } + close(release) +} diff --git a/internal/link/declare.go b/internal/link/declare.go index 63eca97d..be53528b 100644 --- a/internal/link/declare.go +++ b/internal/link/declare.go @@ -41,7 +41,7 @@ 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 } @@ -66,8 +66,16 @@ func Declare(ctx context.Context, bus Bus, signer Signer, node string, return bus.PublishDeclaration(publish, node, body) } -// NearTheLimit is the share of the bus's largest message at which a declaration is said to be near it. -const NearTheLimit = 0.9 +// 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). diff --git a/internal/link/held_asks_test.go b/internal/link/held_asks_test.go new file mode 100644 index 00000000..9119dfcc --- /dev/null +++ b/internal/link/held_asks_test.go @@ -0,0 +1,43 @@ +package link + +import ( + "testing" + "time" +) + +// An ask a machine holds but does not start gets no heartbeat, and the bus hands it to another machine after the ack +// wait, so two machines run one check (issue 1649). + +// A fetch's server-side pull request ends clearly before the fetch does, so the server's answer and any message it could still +// deliver come inside the fetch and never into a buffer nobody reads while the machine builds. +func TestAFetchsPullRequestEndsClearlyBeforeTheFetchDoes(t *testing.T) { + for name, wait := range map[string]time.Duration{"a check queue": checkTakeWait, "the idle wait": takeIdleWait} { + if margin := wait - fetchExpiryOf(wait); margin < minFetchMargin { + t.Errorf("%s: the pull request ends %v before a fetch of %v does; want at least %v", name, margin, wait, minFetchMargin) + } + } +} + +func TestAnAskAnotherBuilderIsRunningIsADuplicateAndYoursIsNot(t *testing.T) { + r := aPullCheck("build-1", 423, "head-4771ee53aa", "trunk-0123456789", false) + run := CheckRunning{Head: r.Ref, Trunk: "trunk-0123456789", Builder: "shanks", Build: "build-1", Started: time.Now().Add(-time.Minute), Seen: time.Now().Add(-4 * time.Second)} + dead := run + dead.Seen = time.Now().Add(-60 * time.Second) + for name, c := range map[string]struct { + running CheckRunning + found bool + me string + dup bool + }{ + "another builder runs this very ask": {run, true, "ace", true}, + "its builder stopped renewing the key a minute ago": {dead, true, "ace", false}, + "this builder runs it itself": {run, true, "shanks", false}, + "nobody runs it": {CheckRunning{}, false, "ace", false}, + "another ask of the same head": {CheckRunning{Head: r.Ref, Trunk: "trunk-0123456789", Builder: "shanks", Build: "build-0"}, true, "ace", false}, + "the same ask at another head": {CheckRunning{Head: "head-other-0000", Trunk: "trunk-0123456789", Builder: "shanks", Build: "build-1"}, true, "ace", false}, + } { + if got := duplicateOf(c.running, c.found, r, c.me) != ""; got != c.dup { + t.Errorf("%s: duplicate = %v; want %v", name, got, c.dup) + } + } +} diff --git a/testdata/beside/CAPTURED b/testdata/beside/CAPTURED index 786b4ff3..7f81fd0d 100644 --- a/testdata/beside/CAPTURED +++ b/testdata/beside/CAPTURED @@ -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 d4deecdb380b69358aae2fdaedcfcc06e1571d42 conformance/events/module-event.json +mesh-sdk 725590e06e75c67b96ce58d015b3056c2d261249 conformance/events/module-event.json To move them, from this repository's root, with the two repositories checked out beside it: diff --git a/vendor/git.novox.be/novox/mesh-sdk/go/declaration/declaration.go b/vendor/git.novox.be/novox/mesh-sdk/go/declaration/declaration.go index 60e5521c..cec208de 100644 --- a/vendor/git.novox.be/novox/mesh-sdk/go/declaration/declaration.go +++ b/vendor/git.novox.be/novox/mesh-sdk/go/declaration/declaration.go @@ -326,7 +326,7 @@ type File struct { // 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 FileBucket, checks what it + // (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. diff --git a/vendor/git.novox.be/novox/mesh-sdk/go/declaration/filereference.go b/vendor/git.novox.be/novox/mesh-sdk/go/declaration/filereference.go index 166022a0..665b0fea 100644 --- a/vendor/git.novox.be/novox/mesh-sdk/go/declaration/filereference.go +++ b/vendor/git.novox.be/novox/mesh-sdk/go/declaration/filereference.go @@ -14,13 +14,38 @@ 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 - // FileBucket is the object bucket of the bus that holds the contents, each under the hex of its sha256 digest - // (ObjectName). The controller alone writes it; a node-engine reads an object by name. - FileBucket = "declaration-files" + // 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-" (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>". diff --git a/vendor/modules.txt b/vendor/modules.txt index c0114ffc..49cf5d54 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -1,4 +1,4 @@ -# git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021144048-d4deecdb380b +# git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021153908-725590e06e75 ## explicit; go 1.26.0 git.novox.be/novox/mesh-sdk/go/asks git.novox.be/novox/mesh-sdk/go/catalogue