O-5: C-S6: stateobjects.Sender and Receive, a reply or an event over the message limit sent as the reference of an object in the sender's store of sent objects (milestone 47/G, issue 1966, ADR 0666 section 16)

Builds: design 47 G §59, §407

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

go/stateobjects/fakes_test.go

@@ -192,6 +192,7 @@ func (m *memory) serve(method string, raw json.RawMessage) (any, error) {
 		Revision uint64          `json:"revision"`
 		Name     string          `json:"name"`
 		Data     []byte          `json:"data"`
+		Object   string          `json:"object"`
 	}
 	_ = json.Unmarshal(raw, &p)
 	switch method {
@@ -214,6 +215,9 @@ func (m *memory) serve(method string, raw json.RawMessage) (any, error) {
 	case buscalls.ObjectsGet:
 		d, err := m.Object(p.Name)
 		return buscalls.ObjectData{Info: buscalls.ObjectInfo{Name: p.Name}, Data: d}, err
+	case buscalls.ObjectsSentGet:
+		d, err := m.Object(p.Object)
+		return buscalls.ObjectData{Info: buscalls.ObjectInfo{Name: p.Object}, Data: d}, err
 	case buscalls.ObjectsDelete:
 		return buscalls.Done{}, memObjects{m}.Delete(p.Name)
 	case buscalls.ObjectsList:

go/stateobjects/send.go

@@ -0,0 +1,93 @@
+package stateobjects
+
+import (
+	"encoding/json"
+	"fmt"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/buscalls"
+)
+
+// MessageLimit is the most a message of the bus carries when the sender does not know its connection's own
+// (`nc.MaxPayload()`), and MessageHeaderRoom what the headers and the subject take of it.
+const (
+	MessageLimit      = 1 << 20
+	MessageHeaderRoom = 8 << 10
+)
+
+// Sender sends the body of a verb's reply or an event: a body over the per-message limit is put in the sender's store of
+// sent objects and its reference is sent instead (ADR 0666 §16). The store is the bus module's `<module>_sent-objects`,
+// granted to the sender alone and read by the callers its seat definition names and the consumers that declare the event;
+// a sent object lives as long as that store's age, which is transport and not kept data.
+type Sender struct {
+	// Objects is the sender's store of sent objects, and Store its name as a reader asks for it.
+	Objects Objects
+	Store   string
+	// Limit is the connection's largest message; zero is MessageLimit.
+	Limit int
+}
+
+// Send is what to send for body: the body itself when it fits, else the reference of the object it was put in as
+// `<label>.<sha256>`. label tells the object's use (the verb or event, the caller or the id).
+func (s Sender) Send(label string, body []byte) ([]byte, error) {
+	limit := s.Limit
+	if limit <= 0 {
+		limit = MessageLimit
+	}
+	if len(body) <= limit-MessageHeaderRoom {
+		if _, isRef := ReadRef(body); isRef {
+			return nil, fmt.Errorf("%s: %w", label, errLooksLikeRef)
+		}
+		return body, nil
+	}
+	if most := s.Objects.MostBytes(); most > 0 && len(body) > most {
+		return nil, fmt.Errorf("%s: a bundle sends at most %d bytes of body, the runtime's channel carrying a message of 16 MiB; this one is %d", label, most, len(body))
+	}
+	ref := Ref{Store: s.Store, Object: objectName(label, digestOf(body)), Size: int64(len(body)), Digest: digestOf(body), Form: FormVersion}
+	if _, err := putObject(s.Objects, ref.Object, body); err != nil {
+		return nil, fmt.Errorf("the object for %s: %w", label, err)
+	}
+	return Wrap(ref), nil
+}
+
+// Fetch reads the object a reference names.
+type Fetch func(Ref) ([]byte, error)
+
+// Receive is the body that was sent: the body itself, or the object its reference names, fetched and checked against the
+// reference's size and digest.
+func Receive(body []byte, fetch Fetch) ([]byte, error) {
+	ref, ok := ReadRef(body)
+	if !ok {
+		return body, nil
+	}
+	data, err := fetch(ref)
+	if err != nil {
+		return nil, fmt.Errorf("the object %s of %s: %w", ref.Object, ref.Store, err)
+	}
+	if int64(len(data)) != ref.Size || digestOf(data) != ref.Digest {
+		return nil, fmt.Errorf("the object %s of %s is not the one its reference names (size or digest differ)", ref.Object, ref.Store)
+	}
+	return data, nil
+}
+
+// BundleFetch reads a sent object through the runtime, which serves it under the standing grant the grants part derives
+// from the manifests (buscalls.ObjectsSentGet).
+func BundleFetch(ref Ref) ([]byte, error) {
+	if ref.Parts > 1 {
+		return nil, fmt.Errorf("the object %s is split over %d objects, which a bundle does not read: a sender in a bundle never splits one", ref.Object, ref.Parts)
+	}
+	raw, err := stdio.BusState(buscalls.ObjectsSentGet, buscalls.ObjectsSentGetRequest{Store: ref.Store, Object: ref.Object, Size: ref.Size, Digest: ref.Digest})
+	if err != nil {
+		return nil, err
+	}
+	var d buscalls.ObjectData
+	if err := json.Unmarshal(raw, &d); err != nil {
+		return nil, fmt.Errorf("the runtime's answer to %s: %w", buscalls.ObjectsSentGet, err)
+	}
+	return d.Data, nil
+}
+
+// ProcessFetch reads a sent object through the reader's own object store.
+func ProcessFetch(o Objects) Fetch {
+	return func(ref Ref) ([]byte, error) { return getObject(o, ref) }
+}

go/stateobjects/send_test.go

@@ -0,0 +1,68 @@
+package stateobjects
+
+import (
+	"bytes"
+	"strings"
+	"testing"
+)
+
+// ADR 0666 §16: a reply or an event over the per-message limit carries the name of an object, and the receiver reads it
+// back whole; one that fits is sent as it is.
+func TestABodyOverTheMessageLimitIsSentAsAReference(t *testing.T) {
+	m := newMemory()
+	s := Sender{Objects: memObjects{m}, Store: "m_sent-objects"}
+	fetch := ProcessFetch(memObjects{m})
+	for _, label := range []string{"reply.list", "event.done"} {
+		body := []byte(`{"text":"` + strings.Repeat("y", 3<<20) + `"}`)
+		sent, err := s.Send(label, body)
+		if err != nil || len(sent) > MessageLimit {
+			t.Fatalf("%s: sent %d bytes, %v", label, len(sent), err)
+		}
+		if _, isRef := ReadRef(sent); !isRef {
+			t.Fatalf("%s: a 3 MiB body was sent whole", label)
+		}
+		got, err := Receive(sent, fetch)
+		if err != nil || !bytes.Equal(got, body) {
+			t.Fatalf("%s: received %d bytes, %v", label, len(got), err)
+		}
+	}
+	small := []byte(`{"a":1}`)
+	if sent, err := s.Send("x", small); err != nil || !bytes.Equal(sent, small) {
+		t.Fatalf("a small body: %s, %v", sent, err)
+	}
+	if got, err := Receive(small, fetch); err != nil || !bytes.Equal(got, small) {
+		t.Fatalf("receive a small body: %s, %v", got, err)
+	}
+	// an object that is not the one the reference names is refused
+	sent, _ := s.Send("reply.again", bytes.Repeat([]byte("z"), 2<<20))
+	ref, _ := ReadRef(sent)
+	m.objects[ref.Object] = []byte("tampered")
+	if _, err := Receive(sent, fetch); err == nil {
+		t.Fatal("a tampered object was accepted")
+	}
+	// a bundle's store takes at most what one runtime message carries
+	bundle := Sender{Objects: BundleObjects{}, Store: "m_sent-objects"}
+	if _, err := bundle.Send("reply.huge", make([]byte, BundleObjectMost+1)); err == nil {
+		t.Fatal("a bundle sent a body its runtime cannot carry")
+	}
+}
+
+// A bundle reads a sent object through the runtime's sent-get.
+func TestABundleReadsASentObjectThroughTheRuntime(t *testing.T) {
+	m := newMemory()
+	body := []byte(`{"text":"` + strings.Repeat("y", 3<<20) + `"}`)
+	sent, err := Sender{Objects: memObjects{m}, Store: "m_sent-objects"}.Send("reply.list", body)
+	if err != nil {
+		t.Fatal(err)
+	}
+	out := runtimeOf(t, m, func() (any, error) {
+		got, err := Receive(sent, BundleFetch)
+		if err != nil || !bytes.Equal(got, body) {
+			t.Errorf("received %d bytes, %v", len(got), err)
+		}
+		return "done", nil
+	})
+	if !strings.Contains(out.(string), "done") {
+		t.Fatalf("the bundle answered %v", out)
+	}
+}

go/stateobjects/values_integration_test.go

@@ -3,7 +3,9 @@
 package stateobjects
 
 import (
+	"bytes"
 	"context"
+	"strings"
 	"testing"
 	"time"
 
@@ -35,5 +37,23 @@ func TestStateObjectsWholeAgainstTheBus(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	sent, err := js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: "m_sent-objects", Storage: jetstream.MemoryStorage, TTL: 7 * 24 * time.Hour})
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Run("a reply and an event over the message limit", func(t *testing.T) {
+		s := Sender{Objects: BusObjects{Store: sent}, Store: "m_sent-objects", Limit: int(nc.MaxPayload())}
+		for _, label := range []string{"reply.list", "event.done"} {
+			body := []byte(`{"text":"` + strings.Repeat("y", 3<<20) + `"}`)
+			wire, err := s.Send(label, body)
+			if err != nil || len(wire) > int(nc.MaxPayload()) {
+				t.Fatalf("%s: sent %d bytes, %v", label, len(wire), err)
+			}
+			got, err := Receive(wire, ProcessFetch(BusObjects{Store: sent}))
+			if err != nil || !bytes.Equal(got, body) {
+				t.Fatalf("%s: received %d bytes, %v", label, len(got), err)
+			}
+		}
+	})
 	t.Run("values of 1x, 10x and 100x the limit", func(t *testing.T) { valueRows(t, OverBus(kv, store, "m_views-objects")) })
 }