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/declaration_files.go b/cmd/mesh-controller/declaration_files.go index c7661b62..e64e3cd2 100644 --- a/cmd/mesh-controller/declaration_files.go +++ b/cmd/mesh-controller/declaration_files.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "sort" + "sync" "git.novox.be/novox/mesh-sdk/go/compose" sdkdecl "git.novox.be/novox/mesh-sdk/go/declaration" @@ -13,6 +14,7 @@ import ( "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). @@ -112,17 +114,20 @@ func namedObjects(body []byte) (map[string]int64, error) { return out, nil } -// objectStore is what putting the objects needs of the bucket. +// 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. - Has(ctx context.Context, name string) (bool, error) - // Put writes the object under that name. + // 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. Refuses on the first it cannot put, naming it, so the declaration is not sent. +// 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 { @@ -130,48 +135,115 @@ func putObjects(ctx context.Context, store objectStore, objects map[string][]byt } sort.Strings(names) for _, name := range names { - have, err := store.Has(ctx, name) + have, marked, err := store.Has(ctx, name) if err != nil { - return fmt.Errorf("whether the object %s is in the bucket %s: %w", name[:12], broker.DeclarationFilesBucket, err) + return fmt.Errorf("whether the object %s is in the bucket: %w", name[:12], err) } - if have { + 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 %s: %w", name[:12], len(objects[name]), - broker.DeclarationFilesBucket, err) + return fmt.Errorf("putting the object %s (%d bytes) in the bucket: %w", name[:12], len(objects[name]), err) } } return nil } -// busObjects is the bucket on the controller's own connection. +// busObjects is a node's bucket on the controller's own connection. type busObjects struct{ store jetstream.ObjectStore } -func openBucket(ctx context.Context, conn *nats.Conn) (busObjects, error) { +// 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 } - store, err := js.ObjectStore(ctx, broker.DeclarationFilesBucket) + store, err := js.ObjectStore(ctx, broker.DeclarationFilesBucketOf(node)) + if errors.Is(err, jetstream.ErrBucketNotFound) { + if err = broker.EnsureDeclarationFiles(ctx, js, node); err == nil { + store, err = js.ObjectStore(ctx, broker.DeclarationFilesBucketOf(node)) + } + } if err != nil { - return busObjects{}, fmt.Errorf("the object bucket %s: %w", broker.DeclarationFilesBucket, err) + return busObjects{}, fmt.Errorf("the object bucket %s: %w", broker.DeclarationFilesBucketOf(node), err) } return busObjects{store}, nil } -func (b busObjects) Has(ctx context.Context, name string) (bool, error) { +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, nil + return false, false, nil case err != nil: - return false, err + return false, false, err } - return !info.Deleted, nil + _, 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 index da9055ba..fec206c0 100644 --- a/cmd/mesh-controller/declaration_files_integration_test.go +++ b/cmd/mesh-controller/declaration_files_integration_test.go @@ -48,11 +48,11 @@ func TestADeclarationOfMoreThanThreeMiBIsSentByReferenceOnARealBus(t *testing.T) if err != nil { t.Fatal(err) } - if err := broker.EnsureDeclarationFiles(ctx, api); err != nil { + 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); err != nil { + if err := broker.EnsureDeclarationFiles(ctx, api, "novox"); err != nil { t.Fatalf("asserting the bucket again: %v", err) } @@ -74,7 +74,7 @@ func TestADeclarationOfMoreThanThreeMiBIsSentByReferenceOnARealBus(t *testing.T) t.Fatalf("the inline declaration of more than 3 MiB: %v", err) } - bucket, err := openBucket(ctx, conn) + bucket, err := openBucket(ctx, conn, "novox") if err != nil { t.Fatal(err) } @@ -101,11 +101,11 @@ func TestADeclarationOfMoreThanThreeMiBIsSentByReferenceOnARealBus(t *testing.T) if err != nil || len(latest) != 1 { t.Fatalf("the latest declarations: %v %v", len(latest), err) } - named, err := namedByLatest(latest) + named, err := namedByLatest("novox", latest["novox"]) if err != nil || len(named) != 3 { t.Fatalf("what the latest declaration names: %v %v", named, err) } - store, err := api.ObjectStore(ctx, broker.DeclarationFilesBucket) + store, err := api.ObjectStore(ctx, broker.DeclarationFilesBucketOf("novox")) if err != nil { t.Fatal(err) } diff --git a/cmd/mesh-controller/declaration_files_test.go b/cmd/mesh-controller/declaration_files_test.go index 6b31ba6a..728ad11e 100644 --- a/cmd/mesh-controller/declaration_files_test.go +++ b/cmd/mesh-controller/declaration_files_test.go @@ -13,6 +13,7 @@ import ( sdkdecl "git.novox.be/novox/mesh-sdk/go/declaration" + "github.com/novox/mesh-controller/internal/conditions" "github.com/novox/mesh-controller/internal/link" ) @@ -202,14 +203,15 @@ func TestAFileOfBytesTravelsByDigestOnItsDecodedLength(t *testing.T) { } type memObjects struct { - have map[string][]byte - puts []string - fail map[string]error + have map[string][]byte + marked map[string]bool + puts []string + fail map[string]error } -func (m *memObjects) Has(_ context.Context, name string) (bool, error) { +func (m *memObjects) Has(_ context.Context, name string) (bool, bool, error) { _, ok := m.have[name] - return ok, nil + return ok, m.marked[name], nil } func (m *memObjects) Put(_ context.Context, name string, c []byte) error { if err := m.fail[name]; err != nil { @@ -217,6 +219,7 @@ func (m *memObjects) Put(_ context.Context, name string, c []byte) error { } m.puts = append(m.puts, name) m.have[name] = c + delete(m.marked, name) return nil } @@ -227,6 +230,12 @@ func TestObjectsAreInTheBucketBeforeTheDeclarationIsSent(t *testing.T) { // 1696 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") { @@ -303,10 +312,13 @@ func (b *sweepBucketFake) ClearUnnamed(_ context.Context, n string) error { b.objects[n] = time.Time{} return nil } -func (b *sweepBucketFake) Delete(_ context.Context, n string) error { +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 nil + 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 @@ -348,12 +360,15 @@ 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(map[string][]byte{"novox": msg, "ace": []byte(`{"declaration":"` + "e30=" + `","signature":""}`)}) + 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 _, err := namedByLatest(map[string][]byte{"novox": []byte("not json")}); err == nil { + 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") } } @@ -379,3 +394,76 @@ func TestTheControlNodesDeclarationInTheReferenceFormStaysUnderHalfTheBusLimit(t } 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") + } +} diff --git a/cmd/mesh-controller/declaration_sweep.go b/cmd/mesh-controller/declaration_sweep.go index c0f11f1d..dcfd6f92 100644 --- a/cmd/mesh-controller/declaration_sweep.go +++ b/cmd/mesh-controller/declaration_sweep.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "sort" + "strings" "time" "github.com/nats-io/nats.go" @@ -20,7 +21,8 @@ import ( // 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. Objects are immutable and regenerable from the +// 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. @@ -41,7 +43,9 @@ 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(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. @@ -72,10 +76,15 @@ func sweepDeclarationFiles(ctx context.Context, bucket sweptBucket, named map[st } r.Marked++ case now.Sub(o.UnnamedSince) >= broker.DeclarationFilesKeptFor: - if err := bucket.Delete(ctx, o.Name); err != nil { + gone, err := bucket.Delete(ctx, o.Name, o.UnnamedSince) + if err != nil { return r, fmt.Errorf("the object %s: %w", o.Name[:12], err) } - r.Removed++ + if gone { + r.Removed++ + } else { + r.Kept++ + } default: r.Kept++ } @@ -83,22 +92,20 @@ func sweepDeclarationFiles(ctx context.Context, bucket sweptBucket, named map[st return r, nil } -// namedByLatest is the name of every object any of these latest declarations names. A body that cannot be read is an error, never -// "names none": an object a node still needs would be marked for removal. -func namedByLatest(bodies map[string][]byte) (map[string]bool, error) { - named := map[string]bool{} - for node, signed := range bodies { - 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) - } - for name := range names { - named[name] = true - } +// 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 } @@ -169,8 +176,18 @@ 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) error { - return b.store.Delete(ctx, name) +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. @@ -199,40 +216,54 @@ func sweepingDeclarationFiles(ctx context.Context, inv *inventory.Inventory, con } } +// 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 { - working, cancel := context.WithTimeout(ctx, 2*time.Minute) - defer cancel() - nodes, err := inv.Nodes(working) + nodes, err := inv.Nodes(ctx) if err != nil { return fmt.Sprintf("the nodes cannot be read: %v", err) } - names := make([]string, 0, len(nodes)) + var problems []string + var total filesSweep for _, n := range nodes { - names = append(names, n.Name) + 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)) + } } - bodies, err := latestDeclarations(working, conn, names) - if err != nil { - return err.Error() - } - named, err := namedByLatest(bodies) - if err != nil { - return err.Error() + 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 err.Error() + return filesSweep{}, err } - store, err := js.ObjectStore(working, broker.DeclarationFilesBucket) - if err != nil { - return fmt.Sprintf("the bucket cannot be read: %v", err) - } - r, err := sweepDeclarationFiles(working, busSweptBucket{store}, named, time.Now()) - if r.Removed+r.Marked+r.Cleared > 0 { - fmt.Printf("the bucket of declaration files: %d object(s) kept, %d found unnamed, %d named again, %d removed\n", - r.Kept, r.Marked, r.Cleared, r.Removed) + store, err := js.ObjectStore(working, broker.DeclarationFilesBucketOf(node)) + if errors.Is(err, jetstream.ErrBucketNotFound) { + return filesSweep{}, nil // a node never sent a reference has no bucket yet } if err != nil { - return err.Error() + return filesSweep{}, fmt.Errorf("the bucket cannot be read: %w", err) } - return "" + 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 bb059262..e3c12494 100644 --- a/cmd/mesh-controller/push.go +++ b/cmd/mesh-controller/push.go @@ -1067,13 +1067,14 @@ func (b overTheBus) putFiles(ctx context.Context, s readyNode) error { if !ok { return fmt.Errorf("%s: the declaration names files by digest and this bus has no object bucket", s.node) } - store, err := openBucket(ctx, bus.Conn) - if err != nil { - return fmt.Errorf("%s was not sent its declaration: %w: %w", s.node, errFilesNotPut, err) - } putting, cancel := context.WithTimeout(ctx, 30*time.Second) defer cancel() - if err := putObjects(putting, store, objects); err != nil { + 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 @@ -1096,11 +1097,17 @@ func (b overTheBus) grant(ctx context.Context, sending []readyNode) error { func (b overTheBus) declare(ctx context.Context, s readyNode, body []byte) (string, error) { // 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. - if err := b.putFiles(ctx, s); err != nil { + // **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 } - if err := link.Declare(ctx, b.server.Bus(), b.signer, s.node, body, 15*time.Second); err != nil { - 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. diff --git a/internal/broker/controller_buckets.go b/internal/broker/controller_buckets.go index 06eec05a..9dd511de 100644 --- a/internal/broker/controller_buckets.go +++ b/internal/broker/controller_buckets.go @@ -231,10 +231,6 @@ func (j *JetStream) EnsureControllerBuckets() error { if err := EnsureLeaseBucket(ctx, js); err != nil { return err } - // And the object bucket a declaration's large files travel in (novox/hq ADR 0486 §4): it exists whenever the bus does. - if err := EnsureDeclarationFiles(ctx, js); err != nil { - return err - } if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{ Bucket: CallsBucket, Description: "the calls of the mesh's own verbs and what came of each (novox/hq to-be 45 §6, issue " + diff --git a/internal/broker/declaration_files.go b/internal/broker/declaration_files.go index 4bec6122..b5261582 100644 --- a/internal/broker/declaration_files.go +++ b/internal/broker/declaration_files.go @@ -9,81 +9,93 @@ import ( "github.com/nats-io/nats.go/jetstream" ) -// The object bucket a declaration's large files travel in (novox/hq ADR 0486, issues 1696 and 1697). +// 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 it, and a node-engine reads an object by its name** (ADR 0486 §2): its account may -// neither list the bucket nor write it, and no other account has any access. A bucket of the bus's object store is a -// stream, OBJ_, whose messages are the objects' metadata (`$O..M.`) and chunks -// (`$O..C.`). +// **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 bucket's name, which mesh-sdk fixes for the node-engine and the controller alike. +// DeclarationFilesBucket is the prefix of the buckets' names, which mesh-sdk fixes for the node-engine and the controller alike. const DeclarationFilesBucket = declaration.FileBucket -// DeclarationFilesStream is the stream under the bucket. -const DeclarationFilesStream = "OBJ_" + DeclarationFilesBucket +// DeclarationFilesBucketOf is the bucket a node's files are in. One function, so the name of a node's bucket is composed in one +// place. +func DeclarationFilesBucketOf(node string) string { return DeclarationFilesBucket + "-" + node } -// DeclarationFilesMaxBytes is the bucket's size: the controller's objects are immutable and regenerable, and the -// largest declaration it composes holds well under a hundred megabytes of them. +// DeclarationFilesStreamOf is the stream under a node's bucket. +func DeclarationFilesStreamOf(node string) string { return "OBJ_" + DeclarationFilesBucketOf(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 the bucket: every object's metadata -// and chunks. The server's API for the stream is the controller's `$JS.API.>`. -const DeclarationFilesWrites = "$O." + DeclarationFilesBucket + ".>" +// 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 { return "$O." + DeclarationFilesBucketOf(node) + ".>" } -// DeclarationFilesReads is what a node-engine's account may publish to read an object 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 bucket and S its +// 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, every -// object's name and chunks. +// 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 list that is refused. The chunks' -// subject carries the nuid the metadata gives, which only a reader of the metadata knows. +// 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). -func DeclarationFilesReads() []string { +// **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 { + stream, bucket := DeclarationFilesStreamOf(node), DeclarationFilesBucketOf(node) return []string{ - "$JS.API.DIRECT.GET." + DeclarationFilesStream + ".$O." + DeclarationFilesBucket + ".M.*", - "$JS.API.CONSUMER.CREATE." + DeclarationFilesStream + ".*.$O." + DeclarationFilesBucket + ".C.*", - "$JS.API.CONSUMER.DELETE." + DeclarationFilesStream + ".*", - "$JS.FC." + DeclarationFilesStream + ".>", + "$JS.API.DIRECT.GET." + stream + ".$O." + bucket + ".M.*", + "$JS.API.CONSUMER.CREATE." + stream + ".*.$O." + bucket + ".C.*", + "$JS.API.CONSUMER.DELETE." + stream + ".*", + "$JS.FC." + stream + ".>", } } -// EnsureDeclarationFiles creates the bucket if it is absent and brings its options to match. The 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() error { +// 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(), 10*time.Second) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() - return EnsureDeclarationFiles(ctx, js) + return EnsureDeclarationFiles(ctx, js, nodes...) } // EnsureDeclarationFiles is the same on a JetStream the caller holds. -func EnsureDeclarationFiles(ctx context.Context, js jetstream.JetStream) error { - if _, err := js.CreateOrUpdateObjectStore(ctx, DeclarationFilesConfig()); err != nil { - return fmt.Errorf("asserting the object bucket %s: %w", DeclarationFilesBucket, err) +func EnsureDeclarationFiles(ctx context.Context, js jetstream.JetStream, nodes ...string) error { + for _, node := range nodes { + if _, err := js.CreateOrUpdateObjectStore(ctx, DeclarationFilesConfig(node)); err != nil { + return fmt.Errorf("asserting the object bucket %s: %w", DeclarationFilesBucketOf(node), err) + } } return nil } -// DeclarationFilesConfig is the bucket as the controller asserts it. -func DeclarationFilesConfig() jetstream.ObjectStoreConfig { +// DeclarationFilesConfig is a node's bucket as the controller asserts it. +func DeclarationFilesConfig(node string) jetstream.ObjectStoreConfig { return jetstream.ObjectStoreConfig{ - Bucket: DeclarationFilesBucket, - Description: "the contents of a declaration's files over 64 KiB, each under the hex of its sha256 digest " + - "(novox/hq ADR 0486): written by the controller alone, read by a node-engine by name, removed a day after " + - "no node's latest declaration names it", + Bucket: DeclarationFilesBucketOf(node), + 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, } diff --git a/internal/broker/declaration_files_live_test.go b/internal/broker/declaration_files_live_test.go index f45b9f33..414c8357 100644 --- a/internal/broker/declaration_files_live_test.go +++ b/internal/broker/declaration_files_live_test.go @@ -25,13 +25,13 @@ import ( // 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, name string) ([]byte, error) { +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 := "OBJ_" + DeclarationFilesBucket - meta, err := js.GetLastMsg(stream, "$O."+DeclarationFilesBucket+".M."+base64.URLEncoding.EncodeToString([]byte(name)), + stream, bucket := DeclarationFilesStreamOf(node), DeclarationFilesBucketOf(node) + meta, err := js.GetLastMsg(stream, "$O."+bucket+".M."+base64.URLEncoding.EncodeToString([]byte(name)), nats.DirectGet(), nats.Context(ctx)) if err != nil { return nil, err @@ -43,7 +43,7 @@ func nodeEnginesFileObject(ctx context.Context, nc *nats.Conn, name string) ([]b if err := json.Unmarshal(meta.Data, &info); err != nil { return nil, err } - sub, err := js.SubscribeSync("$O."+DeclarationFilesBucket+".C."+info.NUID, nats.OrderedConsumer(), nats.BindStream(stream)) + sub, err := js.SubscribeSync("$O."+bucket+".C."+info.NUID, nats.OrderedConsumer(), nats.BindStream(stream)) if err != nil { return nil, err } @@ -59,16 +59,18 @@ func nodeEnginesFileObject(ctx context.Context, nc *nats.Conn, name string) ([]b return body, nil } -// **The bucket's 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, a node-engine's account reads it 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; another module's account is refused all of them. -func TestNatsANodeEngineReadsAnObjectByNameAndIsRefusedTheListAndTheWrite(t *testing.T) { // 1697/AC2 +// **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"}, Assigned: map[string][]Declared{ + users, err := Users(Records{Nodes: []string{"alpha", "beta"}, Assigned: map[string][]Declared{ "alpha": {{Module: RuntimeModule}, {Module: "other-module"}}, }}) if err != nil { @@ -101,7 +103,7 @@ func TestNatsANodeEngineReadsAnObjectByNameAndIsRefusedTheListAndTheWrite(t *tes } t.Cleanup(s.Shutdown) - ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) defer cancel() connect := func(user string) (*nats.Conn, chan error) { t.Helper() @@ -115,26 +117,29 @@ func TestNatsANodeEngineReadsAnObjectByNameAndIsRefusedTheListAndTheWrite(t *tes return nc, refused } - // The controller makes the bucket (as it does at its start) and writes two objects: a small one and one of - // several chunks, which needs the consumer's flow control. + // 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); err != nil { - t.Fatalf("the controller cannot make %s: %v", DeclarationFilesBucket, err) + if err := EnsureDeclarationFiles(ctx, cjs, "alpha", "beta"); err != nil { + t.Fatalf("the controller cannot make the buckets: %v", err) } - store, err := cjs.ObjectStore(ctx, DeclarationFilesBucket) - if err != nil { - t.Fatal(err) + if err := EnsureDeclarationFiles(ctx, cjs, "alpha", "beta"); err != nil { + t.Fatalf("asserting them again: %v", err) } - put := func(content []byte) string { + put := func(node string, content []byte) string { t.Helper() + store, err := cjs.ObjectStore(ctx, DeclarationFilesBucketOf(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: %v", name[:8], err) + t.Fatalf("the controller cannot write the object %s of %s: %v", name[:8], node, err) } return name } @@ -143,62 +148,71 @@ func TestNatsANodeEngineReadsAnObjectByNameAndIsRefusedTheListAndTheWrite(t *tes for i := range large { large[i] = byte(i*7 + i>>8) } - smallName, largeName := put(small), put(large) + secret := bytes.Repeat([]byte("BETA-SECRET "), 50) + smallName, largeName, betaName := put("alpha", small), put("alpha", large), put("beta", secret) - node, nodeRefused := connect("node.alpha") + alpha, alphaRefused := connect("node.alpha") for name, want := range map[string][]byte{smallName: small, largeName: large} { - got, err := nodeEnginesFileObject(ctx, node, name) + got, err := nodeEnginesFileObject(ctx, alpha, "alpha", name) if err != nil { - t.Fatalf("a node-engine cannot read the object %s by its name: %v", name[:8], err) + 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 := <-nodeRefused: + 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 := DeclarationFilesStreamOf("alpha"), DeclarationFilesBucketOf("alpha") + betaStream, betaBucket := DeclarationFilesStreamOf("beta"), DeclarationFilesBucketOf("beta") listing := []byte(`{"subjects_filter":">"}`) bySeq := []byte(`{"seq":1}`) - refusals := []struct { + type refusal struct { what string do func(nc *nats.Conn) - }{ - {"a list of the bucket (stream info asking for every subject)", func(nc *nats.Conn) { - _, _ = nc.Request("$JS.API.STREAM.INFO."+DeclarationFilesStream, listing, 300*time.Millisecond) - }}, - {"a message get by sequence", func(nc *nats.Conn) { - _, _ = nc.Request("$JS.API.STREAM.MSG.GET."+DeclarationFilesStream, bySeq, 300*time.Millisecond) - }}, - {"a direct get by sequence", func(nc *nats.Conn) { - _, _ = nc.Request("$JS.API.DIRECT.GET."+DeclarationFilesStream, bySeq, 300*time.Millisecond) - }}, - {"a consumer on every object's name", func(nc *nats.Conn) { - js, _ := nc.JetStream(nats.MaxWait(500 * time.Millisecond)) - sub, err := js.SubscribeSync("$O."+DeclarationFilesBucket+".M.>", nats.OrderedConsumer(), nats.BindStream(DeclarationFilesStream)) - if err == nil { - sub.Unsubscribe() - } - }}, - {"a write of an object's metadata", func(nc *nats.Conn) { - _ = nc.Publish("$O."+DeclarationFilesBucket+".M.abc", []byte(`{}`)) - _ = nc.Flush() - }}, - {"a write of a chunk", func(nc *nats.Conn) { - _ = nc.Publish("$O."+DeclarationFilesBucket+".C.abc", []byte("x")) - _ = nc.Flush() - }}, } - refusedWith := func(who string, nc *nats.Conn, ch chan error, what string, do func(*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 } - do(nc) + r.do(nc) _ = nc.Flush() deadline := time.After(3 * time.Second) for { @@ -208,32 +222,70 @@ func TestNatsANodeEngineReadsAnObjectByNameAndIsRefusedTheListAndTheWrite(t *tes return } case <-deadline: - t.Errorf("the server did not refuse %s %s", who, what) + t.Errorf("the server did not refuse %s %s", who, r.what) return } } } - for _, r := range refusals { + // 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.what, r.do) + 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.what, r.do) + refusedWith("another module", otherNC, otherCh, r) } - // And another module cannot read an object by name either. - other, otherCh := connect("alpha.other-module") - refusedWith("another module", other, otherCh, "a read of one object by its name", func(nc *nats.Conn) { - _, _ = nc.Request("$JS.API.DIRECT.GET."+DeclarationFilesStream+".$O."+DeclarationFilesBucket+".M."+ - base64.URLEncoding.EncodeToString([]byte(smallName)), nil, 300*time.Millisecond) - }) - // Nothing was written by the refused: the bucket holds the two objects. - var names int - lister, err := store.List(ctx) + // 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) } - names = len(lister) - if names != 2 { - t.Errorf("the bucket holds %d objects, want the 2 the controller wrote", names) + 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, DeclarationFilesBucketOf(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) + } } } diff --git a/internal/broker/declaration_files_test.go b/internal/broker/declaration_files_test.go index c7a23ff4..0de60147 100644 --- a/internal/broker/declaration_files_test.go +++ b/internal/broker/declaration_files_test.go @@ -6,60 +6,72 @@ import ( "testing" ) -// 1697/AC1: the composed accounts give a node-engine exactly the four reads of one object by its name, the controller the -// one write, and no other principal anything of the bucket. -func TestTheDeclarationBucketIsWrittenByTheControllerAndReadByNameByANodeEngine(t *testing.T) { // 1697/AC1 +// 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, DeclarationFilesStream) || strings.Contains(s, "$O."+DeclarationFilesBucket) { + if strings.Contains(s, "declaration-files") { mine = append(mine, s) } } want := []string{ - "$JS.API.CONSUMER.CREATE.OBJ_declaration-files.*.$O.declaration-files.C.*", - "$JS.API.CONSUMER.DELETE.OBJ_declaration-files.*", - "$JS.API.DIRECT.GET.OBJ_declaration-files.$O.declaration-files.M.*", - "$JS.FC.OBJ_declaration-files.>", + "$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 bucket, want exactly %q", mine, want) + t.Fatalf("a node-engine may publish %q of the buckets, want exactly %q", mine, want) } - // What would make a list or a write: none of it, in any pattern a node holds. (The real server's judgement of - // the wildcard forms is the live test's.) + // 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", - "$JS.API.STREAM.MSG.GET.OBJ_declaration-files", - "$JS.API.DIRECT.GET.OBJ_declaration-files", - "$JS.API.DIRECT.GET.OBJ_declaration-files.$O.declaration-files.M.abc.def", - "$JS.API.CONSUMER.CREATE.OBJ_declaration-files.x.$O.declaration-files.M.YWJj", - "$JS.API.CONSUMER.CREATE.OBJ_declaration-files.x.$O.declaration-files.M.YWJj.more", - "$JS.API.CONSUMER.CREATE.OBJ_declaration-files", - "$O.declaration-files.M.abc", - "$O.declaration-files.C.abc", + "$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 the bucket", 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.$O.declaration-files.M.YWJj", - "$JS.API.CONSUMER.CREATE.OBJ_declaration-files.h4sh.$O.declaration-files.C.NUID", + "$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 by its name", 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"}) + controller, err := PermissionsFor(Principal{Kind: KindController, PasswordHash: "x", FilesFor: []string{"alpha", "beta"}}) if err != nil { t.Fatal(err) } - if !MayPublish(controller, "$O.declaration-files.M.abc") || !MayPublish(controller, "$O.declaration-files.C.nuid") { - t.Error("the controller may not write the bucket") + 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{ @@ -78,3 +90,20 @@ func TestTheDeclarationBucketIsWrittenByTheControllerAndReadByNameByANodeEngine( } } } + +// 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) + } +} diff --git a/internal/broker/nats.go b/internal/broker/nats.go index 50cc64ca..06a511a8 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 @@ -392,12 +396,6 @@ var DegradableGrants = []DegradableGrant{ Reason: "carried only at genesis: the controller composes its own permissions and pushes them to the bus at " + "its first push, which is when it acknowledges a merge (ADR 0407 decision 3, issue 973)", }, - { - Subject: DeclarationFilesWrites, - Reason: "carried only at genesis: the controller composes its own permissions and pushes them to the bus at its " + - "first push, and puts a declaration's large files in the bucket only for a node that reports it reads them, " + - "after that (ADR 0486 §1, §2, issue 1697)", - }, { Subject: "$KV." + MergeWindowBucket + ".>", Reason: "refused: the controller closes a batch's window by a timer of its own for that batch, as the " + @@ -744,7 +742,9 @@ func PermissionsFor(p Principal) (Permissions, error) { } // **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. - pub = append(pub, DeclarationFilesWrites) + for _, node := range p.FilesFor { + pub = append(pub, DeclarationFilesWrites(node)) + } // **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. @@ -846,7 +846,7 @@ func PermissionsFor(p Principal) (Permissions, error) { 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). - pub = append(pub, DeclarationFilesReads()...) + pub = append(pub, DeclarationFilesReads(p.Node)...) // 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 891ce58b..1e2b427d 100644 --- a/internal/broker/testdata/composed.conf +++ b/internal/broker/testdata/composed.conf @@ -29,7 +29,7 @@ accounts { ] users = [ { user: "controller", password: "$2a$11$cccccccccccccccccccccc", permissions: { - publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller-merges.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>", "$KV.SEAT_NODE_BUILD_AGENT_cancelled.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_builds.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_checks-running.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_delivery-merges.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_merge-window.>", "$KV.mesh-controller_predefined.>", "$KV.mesh-controller_rollout-silent.>", "$KV.mesh-controller_rollouts.>", "$O.declaration-files.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>", "mesh.seat.mesh-build-machine.tool.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.plan-moved", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.event.sent", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-build-agent.accept.>", "mesh.seat.node-build-agent.check.>", "mesh.seat.node-build-agent.tool.>", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*", "mesh.seat.operator-channel.tool.tell"] } + publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller-merges.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>", "$KV.SEAT_NODE_BUILD_AGENT_cancelled.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_builds.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_checks-running.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_delivery-merges.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_merge-window.>", "$KV.mesh-controller_predefined.>", "$KV.mesh-controller_rollout-silent.>", "$KV.mesh-controller_rollouts.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>", "mesh.seat.mesh-build-machine.tool.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.plan-moved", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.event.sent", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-build-agent.accept.>", "mesh.seat.node-build-agent.check.>", "mesh.seat.node-build-agent.tool.>", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*", "mesh.seat.operator-channel.tool.tell"] } subscribe: { allow: ["$JS.API.>", "$JS.EVENT.ADVISORY.CONSUMER.DELETED.>", "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>", "$SRV.INFO", "$SRV.INFO.mesh-controller", "$SRV.INFO.mesh-controller.>", "$SRV.PING", "$SRV.PING.mesh-controller", "$SRV.PING.mesh-controller.>", "$SRV.STATS", "$SRV.STATS.mesh-controller", "$SRV.STATS.mesh-controller.>", "_DELIVER.controller", "_DELIVER.controller.>", "_INBOX.controller.>", "mesh.control.>", "mesh.mod.*.event.provisioner.failing", "mesh.mod.*.event.provisioner.recovered", "mesh.mod.*.event.provisioner.retirement", "mesh.mod.gitea.event.pull.merged", "mesh.mod.gitea.event.pull.updated", "mesh.mod.mesh-catalog.event.catching-up", "mesh.mod.mesh-catalog.event.upgraded", "mesh.seat.mesh-build-machine.event.built", "mesh.seat.mesh-controller.tool.>", "mesh.seat.node-build-agent.event.built", "mesh.seat.operator-channel.event.decided.mesh-controller"] } allow_responses: { max: 1, ttl: "1m" } } } @@ -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.CREATE.OBJ_declaration-files.*.$O.declaration-files.C.*", "$JS.API.CONSUMER.DELETE.KV_node-engine_have.>", "$JS.API.CONSUMER.DELETE.OBJ_declaration-files.*", "$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.$O.declaration-files.M.*", "$JS.API.STREAM.INFO.KV_node-engine_have", "$JS.FC.KV_node-engine_have.>", "$JS.FC.OBJ_declaration-files.>", "$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 bc27ede2..06a44fde 100644 --- a/internal/broker/writers.go +++ b/internal/broker/writers.go @@ -222,10 +222,10 @@ var WritersTable = []WriterRow{ // 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 + " (stream " + DeclarationFilesStream + "), an object removed a " + - "day after no node's latest declaration names it", - Others: "a node-engine reads an object by its name and may neither list the bucket nor write it", - Subjects: []string{DeclarationFilesWrites}, Writes: isController}, + 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},