diff --git a/go/stateobjects/process.go b/go/stateobjects/process.go new file mode 100644 index 00000000..d799c27a --- /dev/null +++ b/go/stateobjects/process.go @@ -0,0 +1,126 @@ +package stateobjects + +import ( + "context" + "errors" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/kvlimit" +) + +// callTimeout bounds each call a process makes of its bus. +const callTimeout = 30 * time.Second + +// OverBus is a state a process reaches through its own JetStream, with the object store beside it: a foundation part's +// bucket and its old `objects` section, or a bundle-less part's `-objects`. store names the object store as a +// reader asks for it, written into every reference. The bucket's own limit for one value is the limit of the values. +func OverBus(kv jetstream.KeyValue, objects jetstream.ObjectStore, store string) Values { + limit, cancel := context.WithTimeout(context.Background(), callTimeout) + defer cancel() + return Values{KV: BusKV{KV: kv}, Objects: BusObjects{Store: objects}, Store: store, Limit: int(kvlimit.LimitOf(limit, kv))} +} + +// BusKV is a bucket as a process reaches it. +type BusKV struct{ KV jetstream.KeyValue } + +func ctx() (context.Context, context.CancelFunc) { + return context.WithTimeout(context.Background(), callTimeout) +} + +func (b BusKV) Get(key string) (*Entry, error) { + c, cancel := ctx() + defer cancel() + e, err := b.KV.Get(c, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil, nil + } + if err != nil { + return nil, err + } + if e.Operation() != jetstream.KeyValuePut { + return nil, nil + } + return &Entry{Value: e.Value(), Revision: e.Revision()}, nil +} + +func (b BusKV) Create(key string, value []byte) error { + c, cancel := ctx() + defer cancel() + _, err := kvlimit.Create(c, b.KV, key, value) + if errors.Is(err, jetstream.ErrKeyExists) { + return ErrChanged + } + return err +} + +func (b BusKV) Update(key string, value []byte, revision uint64) error { + c, cancel := ctx() + defer cancel() + _, err := kvlimit.Update(c, b.KV, key, value, revision) + if errors.Is(err, jetstream.ErrKeyRevisionMismatch) { + return ErrChanged + } + return err +} + +func (b BusKV) Delete(key string) error { + c, cancel := ctx() + defer cancel() + return b.KV.Delete(c, key) +} + +func (b BusKV) Keys() ([]string, error) { + c, cancel := ctx() + defer cancel() + keys, err := b.KV.Keys(c) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return nil, nil + } + return keys, err +} + +// BusObjects is an object store as a process reaches it; it takes any size in one put. +type BusObjects struct{ Store jetstream.ObjectStore } + +func (BusObjects) MostBytes() int { return 0 } + +func (b BusObjects) Put(name string, data []byte) error { + c, cancel := ctx() + defer cancel() + _, err := b.Store.PutBytes(c, name, data) + return err +} + +func (b BusObjects) Get(name string) ([]byte, error) { + c, cancel := ctx() + defer cancel() + return b.Store.GetBytes(c, name) +} + +func (b BusObjects) Delete(name string) error { + c, cancel := ctx() + defer cancel() + err := b.Store.Delete(c, name) + if errors.Is(err, jetstream.ErrObjectNotFound) { + return nil + } + return err +} + +func (b BusObjects) List() ([]Object, error) { + c, cancel := ctx() + defer cancel() + l, err := b.Store.List(c) + if errors.Is(err, jetstream.ErrNoObjectsFound) { + return nil, nil + } + out := make([]Object, 0, len(l)) + for _, o := range l { + if !o.Deleted { + out = append(out, Object{Name: o.Name, Size: int64(o.Size), Modified: o.ModTime}) + } + } + return out, err +} diff --git a/go/stateobjects/values_integration_test.go b/go/stateobjects/values_integration_test.go new file mode 100644 index 00000000..be0dad25 --- /dev/null +++ b/go/stateobjects/values_integration_test.go @@ -0,0 +1,39 @@ +//go:build integration + +package stateobjects + +import ( + "context" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/testbus" +) + +// ADR 0666 §16, §25 against the real bus: a process's values of 1x, 10x and 100x the limit, replaced and deleted, and a +// body over the per-message limit sent and received, each read back whole. +func TestStateObjectsWholeAgainstTheBus(t *testing.T) { + nc, err := nats.Connect(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + defer nc.Close() + js, err := jetstream.New(nc) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "m_views", MaxValueSize: 256 * 1024, Storage: jetstream.MemoryStorage}) + if err != nil { + t.Fatal(err) + } + store, err := js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: "m_views-objects", Storage: jetstream.MemoryStorage}) + if err != nil { + t.Fatal(err) + } + t.Run("values of 1x, 10x and 100x the limit", func(t *testing.T) { valueRows(t, OverBus(kv, store, "m_views-objects")) }) +}