diff --git a/go/process/emit.go b/go/process/emit.go new file mode 100644 index 0000000..c905ceb --- /dev/null +++ b/go/process/emit.go @@ -0,0 +1,69 @@ +package process + +import ( + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// Emit publishes an event as this process's module (novox/hq ADR 0436 clause 4), under its module's own +// namespace (mesh.mod..event., the same subject the module's own principal would publish it +// on), with the SDK's own envelope and headers (stdio.go:498-509). A chosen message id +// (EmitOptions.MessageID) is sent as x-event-id, and as the bus's own dedup id (Nats-Msg-Id), so a second +// publish inside the stream's duplicate window is dropped; empty is a random one, as stdio.Emit on main. +func (c *Conn) Emit(eventType string, body any, opts ...stdio.EmitOptions) error { + data, err := json.Marshal(body) + if err != nil { + return err + } + id := "" + for _, o := range opts { + if o.MessageID != "" { + id = o.MessageID + } + } + if id == "" { + id = eventID() + } + msg := nats.NewMsg("mesh.mod." + c.account.Module + ".event." + eventType) + msg.Header.Set("x-event-id", id) + msg.Header.Set("x-source", c.account.Module) + msg.Header.Set("x-node", c.account.Node) + msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano)) + msg.Header.Set("content-type", "application/json") + for _, o := range opts { + if o.CausationID != "" { + msg.Header.Set("x-causation-id", o.CausationID) + } + if o.Schema != "" { + msg.Header.Set("x-schema", o.Schema) + } + } + msg.Data = data + js, err := jetstream.New(c.nc) + if err != nil { + return err + } + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + defer cancel() + _, err = js.PublishMsg(ctx, msg, jetstream.WithMsgID(id)) + return err +} + +// eventID is a random UUID, version 4, as stdio's own eventID (go/stdio.go:515-522) — duplicated rather +// than imported, since stdio's is unexported and this package touches none of stdio's bundle machinery. +func eventID() string { + var b [16]byte + _, _ = rand.Read(b[:]) + b[6] = (b[6] & 0x0f) | 0x40 + b[8] = (b[8] & 0x3f) | 0x80 + h := hex.EncodeToString(b[:]) + return h[0:8] + "-" + h[8:12] + "-" + h[12:16] + "-" + h[16:20] + "-" + h[20:32] +} diff --git a/go/process/emit_integration_test.go b/go/process/emit_integration_test.go new file mode 100644 index 0000000..1663b6e --- /dev/null +++ b/go/process/emit_integration_test.go @@ -0,0 +1,57 @@ +//go:build integration + +package process + +import ( + "testing" + "time" + + "github.com/nats-io/nats.go/jetstream" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// 1188/AC3: given a stream accepting mail.send.sent, when the process emits it twice with the same +// message id inside the duplicate window, then the stream holds one message; with the headers stdio's +// Emit sets (stdio.go:498-509). +func TestEmitTwiceWithTheSameIDIsKeptToOneWithTheSDKsHeaders(t *testing.T) { + url, account := grantedSendProcess(t) + account.Node = "novox" + _, adminJetStream := adminJS(t, url) + stream, err := adminJetStream.CreateStream(ctxFor(t), jetstream.StreamConfig{ + Name: "MAIL_EVENTS", Subjects: []string{"mesh.mod.mesh-mail.event.>"}, Duplicates: time.Minute, + }) + if err != nil { + t.Fatalf("the admin could not make the events stream: %v", err) + } + + conn, err := Connect(account) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + + for i := 0; i < 2; i++ { + if err := conn.Emit("mail.send.sent", map[string]any{"send": "s1"}, stdio.EmitOptions{MessageID: "send-s1"}); err != nil { + t.Fatalf("emit %d: %v", i, err) + } + } + + info, err := stream.Info(ctxFor(t)) + if err != nil { + t.Fatal(err) + } + if info.State.Msgs != 1 { + t.Fatalf("the stream holds %d messages, want exactly 1 (the second publish is a duplicate)", info.State.Msgs) + } + + msg, err := stream.GetMsg(ctxFor(t), 1) + if err != nil { + t.Fatal(err) + } + if msg.Header.Get("x-event-id") != "send-s1" || msg.Header.Get("x-source") != "mesh-mail" || + msg.Header.Get("x-node") != "novox" || msg.Header.Get("content-type") != "application/json" || + msg.Header.Get("x-time") == "" { + t.Errorf("the stored message's headers are not the SDK's: %v", msg.Header) + } +} diff --git a/go/process/fixture_integration_test.go b/go/process/fixture_integration_test.go new file mode 100644 index 0000000..27f8468 --- /dev/null +++ b/go/process/fixture_integration_test.go @@ -0,0 +1,100 @@ +//go:build integration + +package process + +import ( + "context" + "testing" + "time" + + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +// sendUser and sendPassword are mesh-mail's process "send" on novox, as mesh-controller would mint it +// (mesh-controller internal/broker/process_grants_test.go mailSenderProcess, Principal.Username). +const ( + sendUser = "novox.mesh-mail.send" + sendPassword = "the-pw" + adminUser = "admin" + adminPass = "admin-pw" +) + +// sendGrants is the process principal mailSenderProcess's whole publish set, copied from +// mesh-controller internal/broker/process_grants_test.go (TestAProcessIsGrantedExactlyItsBusDeclaration, +// "dumped from a passing run... 16 subjects, correct today"), since this package may not import the +// controller: the exact grant the controller composes for novox.mesh-mail.send. +var sendGrants = []string{ + "$JS.ACK.MAIL_SENDS.novox_mesh-mail_send.>", + "$JS.API.CONSUMER.CREATE.KV_mesh-mail_sends.>", + "$JS.API.CONSUMER.DELETE.KV_mesh-mail_sends.>", + "$JS.API.CONSUMER.INFO.MAIL_SENDS.novox_mesh-mail_send", + "$JS.API.CONSUMER.MSG.NEXT.MAIL_SENDS.novox_mesh-mail_send", + "$JS.API.DIRECT.GET.KV_mesh-mail_sends.>", + "$JS.API.DIRECT.GET.KV_messenger_asks.$KV.messenger_asks.mesh-mail.>", + "$JS.API.STREAM.INFO.KV_mesh-mail_sends", + "$JS.FC.KV_mesh-mail_sends.>", + "$KV.mesh-mail_sends.>", + "mesh.mod.mesh-mail.event.mail.send.failed", + "mesh.mod.mesh-mail.event.mail.send.sent", + "mesh.seat.operator-channel.call.tell.*.novox~mesh-mail~send", + "mesh.seat.operator-channel.call.tell.novox~mesh-mail~send", + "mesh.seat.operator-channel.tool.tell", + "mesh.seat.operator-channel.tool.tell.*", +} + +// grantedSendProcess is a test bus with JetStream, with two users: novox.mesh-mail.send, permissioned to +// exactly sendGrants and its own inbox (nats-server server.Options Users with Permissions, as +// mesh-controller's cmd/mesh-controller/process_live_test.go composes accounts for a real bus — copied as +// literal grants here since this package may not import the controller); and an unrestricted admin for the +// test's own setup (creating streams and buckets, standing in for the holder and the runtime). +func grantedSendProcess(t *testing.T) (url string, account Account) { + t.Helper() + opts := &server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(), + NoLog: true, NoSigs: true, + Users: []*server.User{ + {Username: adminUser, Password: adminPass}, + {Username: sendUser, Password: sendPassword, Permissions: &server.Permissions{ + Publish: &server.SubjectPermission{Allow: sendGrants}, + Subscribe: &server.SubjectPermission{Allow: []string{"_INBOX." + sendUser + ".>"}}, + }}, + }, + } + s, err := server.NewServer(opts) + if err != nil { + t.Fatalf("a granted test bus could not be made: %v", err) + } + go s.Start() + if !s.ReadyForConnections(10 * time.Second) { + s.Shutdown() + t.Fatal("a granted test bus did not come up within 10s") + } + t.Cleanup(func() { + s.Shutdown() + s.WaitForShutdown() + }) + url = s.ClientURL() + return url, Account{URL: url, Node: "novox", Module: "mesh-mail", User: sendUser, Password: sendPassword} +} + +// adminJS is the test's own unrestricted connection to url, for setting up fixtures. +func adminJS(t *testing.T, url string) (*nats.Conn, jetstream.JetStream) { + t.Helper() + nc, err := nats.Connect(url, nats.UserInfo(adminUser, adminPass), nats.CustomInboxPrefix("_INBOX."+adminUser)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + js, err := jetstream.New(nc) + if err != nil { + t.Fatal(err) + } + return nc, js +} + +func ctxFor(t *testing.T) context.Context { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + t.Cleanup(cancel) + return ctx +} diff --git a/go/process/state.go b/go/process/state.go new file mode 100644 index 0000000..1f8b6b8 --- /dev/null +++ b/go/process/state.go @@ -0,0 +1,117 @@ +package process + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "time" + + "github.com/nats-io/nats.go/jetstream" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// StateTimeout bounds a state call (Get, Create, Update) and an Emit (emit.go): a var, not a const, so a +// test whose account is not granted a subject — refused only as an async protocol error with no reply, +// never a reply naming the refusal — need not wait out the full bound (mesh-tools' own ProofTimeout, +// node-tools/internal/bus/seat.go:65, is a var for the same reason). +var StateTimeout = 30 * time.Second + +// State is one of this process's module's states, read and written on this connection (novox/hq ADR 0436 +// clause 4), in the bucket _ (mesh-controller internal/broker/state.go:49) — only as the +// account's grants allow. It satisfies go/asker's Store. +type State struct { + Module string + Name string + // kv binds this state's underlying key-value store: a live connection's own (Conn.State), or, in a + // unit test, a fake that never reaches a bus. + kv func(ctx context.Context) (jetstream.KeyValue, error) +} + +// bucket is this state's name on the bus. +func (s State) bucket() string { return s.Module + "_" + s.Name } + +// State is the named state this connection's module keeps, bound lazily to its bucket on each call so a +// bucket the account is not granted fails at the call that needed it, naming the bucket. +func (c *Conn) State(name string) State { + return State{Module: c.account.Module, Name: name, kv: func(ctx context.Context) (jetstream.KeyValue, error) { + js, err := jetstream.New(c.nc) + if err != nil { + return nil, err + } + return js.KeyValue(ctx, c.account.Module+"_"+name) + }} +} + +func (s State) store(ctx context.Context) (jetstream.KeyValue, error) { + kv, err := s.kv(ctx) + if err != nil { + return nil, fmt.Errorf("%s's state %s (bucket %s): %w", s.Module, s.Name, s.bucket(), err) + } + return kv, nil +} + +// Get is a key's current value, or nil when it has none. +func (s State) Get(key string) (*stdio.StateEntry, error) { + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + defer cancel() + kv, err := s.store(ctx) + if err != nil { + return nil, err + } + entry, err := kv.Get(ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err) + } + return &stdio.StateEntry{Key: key, Value: json.RawMessage(entry.Value()), Revision: entry.Revision()}, nil +} + +// Create writes a key only when it has none, and returns the revision; a key that has one is refused as +// stdio.ErrChanged. +func (s State) Create(key string, value any) (uint64, error) { + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + defer cancel() + kv, err := s.store(ctx) + if err != nil { + return 0, err + } + data, err := json.Marshal(value) + if err != nil { + return 0, err + } + rev, err := kv.Create(ctx, key, data) + if errors.Is(err, jetstream.ErrKeyExists) { + return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged) + } + if err != nil { + return 0, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err) + } + return rev, nil +} + +// Update writes a key only when it is still at revision, and returns the new one; a key changed meanwhile +// is refused as stdio.ErrChanged. +func (s State) Update(key string, value any, revision uint64) (uint64, error) { + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + defer cancel() + kv, err := s.store(ctx) + if err != nil { + return 0, err + } + data, err := json.Marshal(value) + if err != nil { + return 0, err + } + rev, err := kv.Update(ctx, key, data, revision) + if errors.Is(err, jetstream.ErrKeyRevisionMismatch) { + return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged) + } + if err != nil { + return 0, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err) + } + return rev, nil +} diff --git a/go/process/state_integration_test.go b/go/process/state_integration_test.go new file mode 100644 index 0000000..1cb63e9 --- /dev/null +++ b/go/process/state_integration_test.go @@ -0,0 +1,88 @@ +//go:build integration + +package process + +import ( + "errors" + "strings" + "testing" + "time" + + "github.com/nats-io/nats.go/jetstream" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// 1188/AC1: given a process connection to a test bus granted the bucket mesh-mail_sends, when it creates a +// key, reads it, and updates it with the revision it read, then each succeeds and returns the new +// revision; an update with a stale revision fails as stdio.ErrChanged; a create of an existing key fails +// as stdio.ErrChanged. +func TestStateCreateGetUpdateOnAGrantedBucket(t *testing.T) { + url, account := grantedSendProcess(t) + _, adminJetStream := adminJS(t, url) + if _, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_sends"}); err != nil { + t.Fatalf("the admin could not make the bucket: %v", err) + } + + conn, err := Connect(account) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + s := conn.State("sends") + + rev, err := s.Create("key", map[string]any{"v": 1}) + if err != nil { + t.Fatalf("Create on a granted bucket: %v", err) + } + if rev == 0 { + t.Fatal("Create returned no revision") + } + + e, err := s.Get("key") + if err != nil || e == nil { + t.Fatalf("Get after Create: %+v, %v", e, err) + } + if e.Revision != rev || string(e.Value) != `{"v":1}` { + t.Fatalf("Get read back %+v, want revision %d and {\"v\":1}", e, rev) + } + + rev2, err := s.Update("key", map[string]any{"v": 2}, rev) + if err != nil { + t.Fatalf("Update at the revision read: %v", err) + } + if rev2 == rev { + t.Fatal("Update did not return a new revision") + } + + if _, err := s.Update("key", map[string]any{"v": 3}, rev); !errors.Is(err, stdio.ErrChanged) { + t.Fatalf("Update at a stale revision: %v, want stdio.ErrChanged", err) + } + + if _, err := s.Create("key", map[string]any{"v": 4}); !errors.Is(err, stdio.ErrChanged) { + t.Fatalf("Create of an existing key: %v, want stdio.ErrChanged", err) + } +} + +// 1188/AC4: given an account not granted a bucket, when it writes there, then the bus's refusal comes back +// as an error naming the bucket. A denied publish is refused only as an async protocol error with no +// reply (never a reply naming the refusal), so StateTimeout is shortened for this test alone. +func TestStateRefusesAnUngrantedBucketNamingIt(t *testing.T) { + old := StateTimeout + StateTimeout = 2 * time.Second + t.Cleanup(func() { StateTimeout = old }) + + _, account := grantedSendProcess(t) + conn, err := Connect(account) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + + s := conn.State("unrelated") + if _, err := s.Create("key", map[string]any{"v": 1}); err == nil { + t.Fatal("writing to an ungranted bucket was not refused") + } else if !strings.Contains(err.Error(), "mesh-mail_unrelated") { + t.Errorf("the refusal %q does not name the bucket mesh-mail_unrelated", err.Error()) + } +} diff --git a/go/process/state_test.go b/go/process/state_test.go new file mode 100644 index 0000000..542153b --- /dev/null +++ b/go/process/state_test.go @@ -0,0 +1,105 @@ +package process + +import ( + "context" + "errors" + "sync" + "testing" + + "github.com/nats-io/nats.go/jetstream" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/asker" +) + +// fakeEntry is a fakeKV's KeyValueEntry: only Key, Value and Revision are read by State. +type fakeEntry struct { + jetstream.KeyValueEntry + key string + value []byte + rev uint64 +} + +func (e fakeEntry) Key() string { return e.key } +func (e fakeEntry) Value() []byte { return e.value } +func (e fakeEntry) Revision() uint64 { return e.rev } + +// fakeKV is an in-memory jetstream.KeyValue: embedding the nil interface satisfies every method this test +// never calls, so only Get, Create and Update need a body (1188/AC2's "a fake KV"). +type fakeKV struct { + jetstream.KeyValue + mu sync.Mutex + values map[string][]byte + revs map[string]uint64 + next uint64 +} + +func newFakeKV() *fakeKV { return &fakeKV{values: map[string][]byte{}, revs: map[string]uint64{}} } + +func (f *fakeKV) Get(_ context.Context, key string) (jetstream.KeyValueEntry, error) { + f.mu.Lock() + defer f.mu.Unlock() + v, ok := f.values[key] + if !ok { + return nil, jetstream.ErrKeyNotFound + } + return fakeEntry{key: key, value: v, rev: f.revs[key]}, nil +} + +func (f *fakeKV) Create(_ context.Context, key string, value []byte, _ ...jetstream.KVCreateOpt) (uint64, error) { + f.mu.Lock() + defer f.mu.Unlock() + if _, ok := f.values[key]; ok { + return 0, jetstream.ErrKeyExists + } + f.next++ + f.values[key], f.revs[key] = value, f.next + return f.next, nil +} + +func (f *fakeKV) Update(_ context.Context, key string, value []byte, revision uint64) (uint64, error) { + f.mu.Lock() + defer f.mu.Unlock() + if f.revs[key] != revision { + return 0, jetstream.ErrKeyRevisionMismatch + } + f.next++ + f.values[key], f.revs[key] = value, f.next + return f.next, nil +} + +func stateOn(kv jetstream.KeyValue) State { + return State{Module: "mesh-mail", Name: "sends", kv: func(context.Context) (jetstream.KeyValue, error) { return kv, nil }} +} + +// 1188/AC2: given the process connection's state type, when used as go/asker's Store, then it compiles and +// behaves as the interface says, against a fake KV. +func TestStateSatisfiesAskerStore(t *testing.T) { + var _ asker.Store = State{} + + s := stateOn(newFakeKV()) + if e, err := s.Get("ask.1"); err != nil || e != nil { + t.Fatalf("Get of an absent key: %v, %v", e, err) + } + rev, err := s.Create("ask.1", map[string]any{"a": 1}) + if err != nil { + t.Fatalf("Create: %v", err) + } + e, err := s.Get("ask.1") + if err != nil || e == nil || e.Revision != rev { + t.Fatalf("Get after Create: %+v, %v", e, err) + } + if _, err := s.Create("ask.1", map[string]any{"a": 2}); !errors.Is(err, stdio.ErrChanged) { + t.Fatalf("Create of an existing key: %v, want stdio.ErrChanged", err) + } + rev2, err := s.Update("ask.1", map[string]any{"a": 3}, rev) + if err != nil { + t.Fatalf("Update at the revision read: %v", err) + } + if _, err := s.Update("ask.1", map[string]any{"a": 4}, rev); !errors.Is(err, stdio.ErrChanged) { + t.Fatalf("Update at a stale revision: %v, want stdio.ErrChanged", err) + } + if rev2 == rev { + t.Fatal("Update did not advance the revision") + } +}