O-3: C-S4: stateobjects.OverBus, the same values of any size through a process's own JetStream, tested against the bus (milestone 47/G, issue 1966, ADR 0666 section 16)

Builds: design 47 G §58, §407

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ

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 `<state>-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
+}

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")) })
+}