novox/mesh-catalog
9
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

bus5 B6 (A-109): snapshot-get's chunks go on a core publish to any caller; no grant check

Browse Source 
grantedBy is dropped: any caller of snapshot-get gets the chunks, a protected snapshot never.
The chunks are a transfer on the core bus, published with a core publish and kept on no stream.

Builds: design 47 G A-109 (G-20-lead-answers), §681

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
jschoubben
2026-11-11 23:57:09 +01:00
parent 44cb7ad449
commit 605c583ab4
4 changed files with 12 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/nats/cmd/nats-tools/keep.go
+1 -1
View File
@@ -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)
}
modules/nats/cmd/nats-tools/keeper_integration_test.go
+1 -5
View File
@@ -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)
modules/nats/cmd/nats-tools/snapshots.go
+7 -37
View File
@@ -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 {
modules/nats/cmd/nats-tools/snapshots_test.go
+3 -11
View File
@@ -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)