diff --git a/modules/nats/cmd/nats-tools/keep.go b/modules/nats/cmd/nats-tools/keep.go index 483de4537..37c18e506 100644 --- a/modules/nats/cmd/nats-tools/keep.go +++ b/modules/nats/cmd/nats-tools/keep.go @@ -281,7 +281,7 @@ func runKeeper(ctx context.Context, getenv func(string) string, logOut io.Writer SnapshotKeepDays: int(number(getenv, "MESH_NATS_SNAPSHOT_KEEP_DAYS", keeper.DefaultSnapshotKeepDays)), } // The process answers the seat mesh-broker's snapshot verbs itself (design 47 G ยง681, A-107), with the caller the bus proved. - stopAnswers, err := conn.AnswerSeat(meshbroker.Seat, snapshotAnswers(k, heldRequest(k.Store), keeperSnapshots(js, nc, account, logger.Printf))) + stopAnswers, err := conn.AnswerSeat(meshbroker.Seat, snapshotAnswers(k, heldRequest(k.Store), keeperSnapshots(nc, logger.Printf))) if err != nil { return fmt.Errorf("the seat %s's snapshot verbs cannot be answered: %w", meshbroker.Seat, err) } diff --git a/modules/nats/cmd/nats-tools/keeper_integration_test.go b/modules/nats/cmd/nats-tools/keeper_integration_test.go index f0031effe..bb06a2a05 100644 --- a/modules/nats/cmd/nats-tools/keeper_integration_test.go +++ b/modules/nats/cmd/nats-tools/keeper_integration_test.go @@ -451,11 +451,7 @@ func TestTheBusModuleRunsWhole(t *testing.T) { if rec.Path == "" || rec.Digest == "" || rec.Protected { t.Fatalf("the drop's record holds no snapshot: %+v", rec) } - if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "CHUNKS", Subjects: []string{"mesh.seat.mesh-broker.chunk.>"}}); err != nil { - t.Fatal(err) - } - world := keeperSnapshots(js, nc, account, t.Logf) - world.Granted = func(string, string) bool { return true } + world := keeperSnapshots(nc, t.Logf) stopAnswering, err := pconn.AnswerSeat(meshbroker.Seat, snapshotAnswers(k, heldRequest(store), world)) if err != nil { t.Fatal(err) diff --git a/modules/nats/cmd/nats-tools/snapshots.go b/modules/nats/cmd/nats-tools/snapshots.go index f1266342c..01f2e03f8 100644 --- a/modules/nats/cmd/nats-tools/snapshots.go +++ b/modules/nats/cmd/nats-tools/snapshots.go @@ -1,7 +1,6 @@ package main import ( - "context" "crypto/sha256" "encoding/hex" "encoding/json" @@ -17,7 +16,6 @@ import ( "time" "github.com/nats-io/nats.go" - "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" stdio "git.novox.be/novox/mesh-sdk/go" @@ -68,8 +66,6 @@ type snapshotWorld struct { // Publish says one chunk on its subject, the process's own publish; Hear subscribes to the acknowledgements, once. Publish func(subject string, body any, id string) error Hear func(ack func(subject string)) error - // Granted says whether the caller may hear the chunks and acknowledge them (the bus's users, nats_user_can). - Granted func(caller, call string) bool Logf func(string, ...any) // Within is how long an acknowledgement is waited for. Within time.Duration @@ -80,43 +76,17 @@ type snapshotWorld struct { waiting map[string]chan struct{} } -// grantedBy is whether the bus's users let caller hear a send's chunks and acknowledge them. -func grantedBy(caller, call string) bool { - ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) - defer cancel() - conf, err := ReadUsers(ctx) - if err != nil { - return false - } - a, err := UserCan(conf, caller, "subscribe", ChunkSubject(call, 1)) - if err != nil || a["allowed"] != true { - return false - } - a, err = UserCan(conf, caller, "publish", AckSubject(call, 1)) - return err == nil && a["allowed"] == true -} - -// keeperSnapshots is the world as the keeper's process reaches it: its own JetStream publish of a chunk, as the seat's events are -// said, and its own core subscription for the acknowledgements, answered at once since the caller asks them as requests. -func keeperSnapshots(js jetstream.JetStream, nc *nats.Conn, a process.Account, logf func(string, ...any)) *snapshotWorld { +// keeperSnapshots is the world as the keeper's process reaches it: its own core publish of a chunk and its own core subscription for the acknowledgements, answered at once since the caller asks them as requests. +func keeperSnapshots(nc *nats.Conn, logf func(string, ...any)) *snapshotWorld { return &snapshotWorld{ Open: func(path string) (io.ReadCloser, error) { return os.Open(path) }, - Publish: func(subject string, body any, id string) error { + // A chunk is a transfer on the core bus, kept on no stream (A-109). + Publish: func(subject string, body any, _ string) error { data, err := json.Marshal(body) if err != nil { return err } - msg := nats.NewMsg(subject) - msg.Header.Set("x-event-id", a.Module+"."+id) - msg.Header.Set("x-source", a.Module) - msg.Header.Set("x-node", a.Node) - msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano)) - msg.Header.Set("content-type", "application/json") - msg.Data = data - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() - _, err = js.PublishMsg(ctx, msg, jetstream.WithMsgID(a.Module+"."+id)) - return err + return nc.Publish(subject, data) }, Hear: func(ack func(subject string)) error { _, err := nc.Subscribe(AckPattern, func(m *nats.Msg) { @@ -127,7 +97,7 @@ func keeperSnapshots(js jetstream.JetStream, nc *nats.Conn, a process.Account, l }) return err }, - Granted: grantedBy, Logf: logf, Within: AckWithin, + Logf: logf, Within: AckWithin, } } @@ -204,7 +174,7 @@ func (w *snapshotWorld) get(held func(id string) (*keeper.Request, error), calle at := rec.At out := meshbroker.SnapshotGetAnswer{Request: rec.ID, Module: rec.Module, Local: rec.Local, Stream: keeper.BusName(rec.Module, rec.Local), Size: size, Digest: digest, At: &at} - if size == 0 || !w.Granted(caller, call) { + if size == 0 { return out, nil } if err := w.hearAcks(); err != nil { diff --git a/modules/nats/cmd/nats-tools/snapshots_test.go b/modules/nats/cmd/nats-tools/snapshots_test.go index 96a7ee8df..acd4ab909 100644 --- a/modules/nats/cmd/nats-tools/snapshots_test.go +++ b/modules/nats/cmd/nats-tools/snapshots_test.go @@ -98,7 +98,6 @@ func TestTheKeepersProcessAnswersTheSnapshotVerbs(t *testing.T) { var journal []string var mu sync.Mutex locked := func(f func()) { mu.Lock(); defer mu.Unlock(); f() } - granted := true w := &snapshotWorld{ Open: func(path string) (io.ReadCloser, error) { b, ok := files[path] @@ -117,9 +116,8 @@ func TestTheKeepersProcessAnswersTheSnapshotVerbs(t *testing.T) { ackHandler = ack return nil }, - Granted: func(string, string) bool { return granted }, - Logf: func(f string, a ...any) { locked(func() { journal = append(journal, fmt.Sprintf(f, a...)) }) }, - Within: time.Millisecond, + Logf: func(f string, a ...any) { locked(func() { journal = append(journal, fmt.Sprintf(f, a...)) }) }, + Within: time.Millisecond, } store := &mapStore{m: map[string][]byte{}} snapFile := t.TempDir() + "/rm.tar" @@ -159,13 +157,7 @@ func TestTheKeepersProcessAnswersTheSnapshotVerbs(t *testing.T) { t.Errorf("%s: answered %+v, want %q and no chunk", row.request, got, row.want) } } - // A caller granted no chunk subjects is answered the fields alone. - granted = false - if got := get("r1", call); got.Error != "" || got.Chunks != 0 || got.Digest != digest || got.Stream != "m_drafts" || got.Size != int64(len(file)) || len(published) != 0 { - t.Errorf("an ungranted caller: %+v, published %d", got, len(published)) - } - // A granted one hears each chunk after the answer, the next only once the last is acknowledged. - granted = true + // Any caller hears each chunk hears each chunk after the answer, the next only once the last is acknowledged. got := get("r1", call) if got.Chunks != 3 { t.Fatalf("answered %+v: 2.5 chunks of 512 KiB are three", got)