Give each node a bucket of its own for the files that travel by digest, and keep the sweep from removing an object a send relies on (issues 1696, 1697; review of ADR 0486 step 3)

A literal * in a published subject matches the grant's *, so in one shared bucket a node could read every node's files. Each node now
has declaration-files-<node>, which its account reads by name and the controller writes; the sweep judges each bucket against that
node's latest declaration, under the node's lock with the send, and an object the sweep marked is put again by the next send. Putting
that fails send after send raises a condition.

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)
+		}
+	}
+}