Let the operator's hand path replace a file over 64 KiB in a node's object bucket (issue 1945; the review of A1, H1)

The object store reads an object's metadata and purges its old chunks when it replaces one, so the hand path is
granted the direct and message gets and the purge of each node's object stream; a test on a bus running the composed
user list walks the hand path end to end.

internal/broker/composed.go

@@ -128,7 +128,10 @@ func HandPathWrites(nodes []string) ([]string, error) {
 		if err != nil {
 			return nil, err
 		}
-		out = append(out, files, "$JS.API.STREAM.INFO.OBJ_declaration-files-"+node)
+		stream := "OBJ_" + DeclarationFilesBucket + node
+		// The object store's put reads an object's metadata already there and, replacing it, purges its old chunks.
+		out = append(out, files, "$JS.API.STREAM.INFO."+stream, "$JS.API.DIRECT.GET."+stream+".>",
+			"$JS.API.STREAM.MSG.GET."+stream, "$JS.API.STREAM.PURGE."+stream)
 	}
 	return out, nil
 }

internal/broker/composed_live_test.go

@@ -0,0 +1,142 @@
+//go:build integration
+
+package broker
+
+import (
+	"bytes"
+	"context"
+	"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"
+)
+
+// 1945 (the review of A1, H1): on a bus running the user list this controller composes, the operator's account (a
+// person who may invoke every tool) walks release 6's hand path end to end: it writes a node's want, that node's
+// composition and its digest, and stores a file over 64 KiB in the node's object bucket and replaces it; a person with
+// fewer tools may do none of it.
+func TestNatsTheOperatorWritesTheHandPathOnTheComposedUserList(t *testing.T) {
+	hash, err := bcrypt.GenerateFromPassword([]byte("pw"), bcrypt.MinCost)
+	if err != nil {
+		t.Fatal(err)
+	}
+	users, err := Users(Records{Nodes: []string{"ace"},
+		People: map[string][]string{"jochen": {"*"}, "guest": {"mesh-controller.status"}}})
+	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(), 60*time.Second)
+	defer cancel()
+	connect := func(user string) jetstream.JetStream {
+		t.Helper()
+		nc, err := nats.Connect(s.ClientURL(), nats.UserInfo(user, "pw"), nats.CustomInboxPrefix("_INBOX."+user))
+		if err != nil {
+			t.Fatalf("%s cannot connect: %v", user, err)
+		}
+		t.Cleanup(nc.Close)
+		js, err := jetstream.New(nc)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return js
+	}
+
+	// The buckets as the controller and the assigner make them.
+	cjs := connect(ControllerName)
+	if err := EnsureDeclarationFiles(ctx, cjs, "ace"); err != nil {
+		t.Fatal(err)
+	}
+	if err := EnsureComposedDigests(ctx, cjs); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := cjs.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "composed-ace", History: 5}); err != nil {
+		t.Fatal(err)
+	}
+	// The assigner's bucket, made here by the controller's account, which may create any stream.
+	if _, err := cjs.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "mesh-delivery-assigner_wants", History: 20}); err != nil {
+		t.Fatal(err)
+	}
+
+	short := func(c context.Context) (context.Context, context.CancelFunc) { return context.WithTimeout(c, 5*time.Second) }
+	for _, who := range []struct {
+		user string
+		may  bool
+	}{{"person.jochen", true}, {"person.guest", false}} {
+		js := connect(who.user)
+		try := func(what string, act func(context.Context) error) {
+			t.Helper()
+			c, stop := short(ctx)
+			defer stop()
+			err := act(c)
+			if (err == nil) != who.may {
+				t.Errorf("%s: %s: %v (may: %v)", who.user, what, err, who.may)
+			}
+		}
+		try("write ace's want", func(c context.Context) error {
+			kv, err := js.KeyValue(c, "mesh-delivery-assigner_wants")
+			if err != nil {
+				return err
+			}
+			_, err = kv.Put(c, "want.ace.mesh-controller", []byte(`{"v":1}`))
+			return err
+		})
+		try("write ace's composition", func(c context.Context) error {
+			kv, err := js.KeyValue(c, "composed-ace")
+			if err != nil {
+				return err
+			}
+			_, err = kv.Put(c, "ace.mesh-controller.aaaa", []byte(`{"v":1}`))
+			return err
+		})
+		try("write its digest", func(c context.Context) error {
+			kv, err := js.KeyValue(c, ControllerComposedBucket)
+			if err != nil {
+				return err
+			}
+			_, err = kv.Put(c, "ace.mesh-controller.aaaa", []byte(`{"v":1}`))
+			return err
+		})
+		try("store and replace a file over 64 KiB", func(c context.Context) error {
+			store, err := js.ObjectStore(c, "declaration-files-ace")
+			if err != nil {
+				return err
+			}
+			body := bytes.Repeat([]byte("x"), 200<<10)
+			if _, err := store.PutBytes(c, "f00d", body); err != nil {
+				return err
+			}
+			_, err = store.PutBytes(c, "f00d", append(body, 'y'))
+			return err
+		})
+	}
+}

internal/broker/composed_test.go

@@ -76,7 +76,9 @@ func TestTheOperatorWritesTheHandPathAheadOfTheSwitch(t *testing.T) {
 			t.Fatal(err)
 		}
 		hand := []string{"$KV.mesh-delivery-assigner_wants.want.>", "$KV.composed-ace.>", "$KV.composed-novox.>",
-			"$KV.mesh-controller_composed.>", "$O.declaration-files-ace.>"}
+			"$KV.mesh-controller_composed.>", "$O.declaration-files-ace.>",
+			"$JS.API.DIRECT.GET.OBJ_declaration-files-ace.>", "$JS.API.STREAM.MSG.GET.OBJ_declaration-files-ace",
+			"$JS.API.STREAM.PURGE.OBJ_declaration-files-ace"}
 		for _, s := range hand {
 			if got := slices.Contains(perms.Publish, s); got != (u.Module == "jochen") {
 				t.Errorf("%s may publish %s: %v", u.Module, s, got)

internal/broker/log_test.go

@@ -206,8 +206,11 @@ func TestNoPathDeletesALogOrABucket(t *testing.T) {
 			})
 			for _, lit := range stringsIn(f) {
 				if strings.Contains(lit, "$JS.API.STREAM.DELETE") || strings.Contains(lit, "$JS.API.STREAM.PURGE") {
-					// Only the writers table names it, as a subject no one but the controller may publish.
-					if !strings.HasSuffix(filepath.ToSlash(path), "internal/broker/writers.go") {
+					// Only the writers table names it, as a subject no one but the controller may publish; and the operator's
+					// release 6 hand path, whose object store purges the chunks of an object it replaces in a node's
+					// object bucket (novox/hq ADR 0539 clause 10): a grant, never a call of this repository.
+					if !strings.HasSuffix(filepath.ToSlash(path), "internal/broker/writers.go") &&
+						!strings.HasSuffix(filepath.ToSlash(path), "internal/broker/composed.go") {
 						found = append(found, path+": "+lit)
 					}
 				}