Delete the push and the declaration: the send stub and node holds, link declare and its order, the sent readers, and the sweep judged by compositions (release 6, issue 1945 B8)

cmd/mesh-controller/declaration_sweep.go

@@ -9,23 +9,22 @@ import (
 	"strings"
 	"time"
 
+	sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
 	"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);
+// **An object is removed a day after no composition of the node names it** (novox/hq ADR 0486 §3, ADR 0539 §4, issue 1696), so a
+// node that is asleep and applies its wants later still finds every object they name. What a node's compositions name is read
+// from its bucket `composed-<node>`, where the controller keeps each module's five newest fingerprints and the node's own: nothing
+// else is kept of them. 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.
+// node's compositions 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.
+// declarationUnnamed is the key of an object's metadata holding when it was first found named by no composition of the node.
 const declarationUnnamed = "unnamed-since"
 
 // sweepEvery is how often the bucket is judged: the removal is exact to within this.
@@ -92,41 +91,62 @@ func sweepDeclarationFiles(ctx context.Context, bucket sweptBucket, named map[st
 	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
+// namedByCompositions is the name of every object the compositions of a node name: the files over 64 KiB each carries by
+// digest (ADR 0539 §4). A composition that cannot be read is an error, never "names none": an object the node still needs
+// would be marked for removal.
+func namedByCompositions(node string, values [][]byte) (map[string]bool, error) {
+	named := map[string]bool{}
+	for _, value := range values {
+		c, err := sdkdelivery.DecodeComposed(value)
+		if err != nil {
+			return nil, fmt.Errorf("a composition of %s is not readable: %w", node, err)
+		}
+		body, err := json.Marshal(map[string]any{"resources": c.Resources})
+		if err != nil {
+			return nil, err
+		}
+		names, err := namedObjects(body)
+		if err != nil {
+			return nil, fmt.Errorf("a composition of %s is not readable: %w", node, err)
+		}
+		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))
+// keptCompositionsOf are the values of every composition in a node's bucket `composed-<node>`: what the controller keeps there
+// are each module's five newest fingerprints and the node's own (ADR 0539 §4). A node with no bucket has none.
+func keptCompositionsOf(ctx context.Context, js jetstream.JetStream, node string) ([][]byte, error) {
+	bucket, err := sdkdelivery.ComposedBucketFor(node)
 	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) {
+	kv, err := js.KeyValue(ctx, bucket)
+	if errors.Is(err, jetstream.ErrBucketNotFound) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, fmt.Errorf("the compositions of %s: %w", node, err)
+	}
+	keys, err := kv.Keys(ctx)
+	if errors.Is(err, jetstream.ErrNoKeysFound) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, fmt.Errorf("the compositions of %s: %w", node, err)
+	}
+	var out [][]byte
+	for _, key := range keys {
+		e, err := kv.Get(ctx, key)
+		if errors.Is(err, jetstream.ErrKeyNotFound) {
 			continue
 		}
 		if err != nil {
-			return nil, fmt.Errorf("the latest declaration of %s: %w", node, err)
+			return nil, fmt.Errorf("the composition %s: %w", key, err)
 		}
-		out[node] = msg.Data
+		out = append(out, e.Value())
 	}
 	return out, nil
 }
@@ -259,15 +279,13 @@ func sweepNode(ctx context.Context, conn *nats.Conn, node string) (filesSweep, e
 		return filesSweep{}, fmt.Errorf("the bucket cannot be read: %w", err)
 	}
 	defer lockFilesOf(node)()
-	latest, err := latestDeclarations(working, conn, []string{node})
+	values, err := keptCompositionsOf(working, js, 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
-		}
+	named, err := namedByCompositions(node, values)
+	if err != nil {
+		return filesSweep{}, err
 	}
 	return sweepDeclarationFiles(working, busSweptBucket{store}, named, time.Now())
 }