feat(plugin): implement chunked body fetching and error handling for …

custom_host_call.go

@@ -45,7 +45,7 @@ func CallHostFunction(hostFn CustomHostImport, request any, response any) error
 	resPtr := int32(uint32(outputBuf[0]) | uint32(outputBuf[1])<<8 | uint32(outputBuf[2])<<16 | uint32(outputBuf[3])<<24)
 	resLen := int32(uint32(outputBuf[4]) | uint32(outputBuf[5])<<8 | uint32(outputBuf[6])<<16 | uint32(outputBuf[7])<<24)
 	if resLen == 0 {
-		return fmt.Errorf("plugin: custom host call: empty response from host")
+		return errEmptyHostResponse("custom host call")
 	}
 	resBytes := append([]byte(nil), wasmSlice(resPtr, resLen)...)
 	wasmResetAllocator()

fetch_chunks.go

@@ -0,0 +1,84 @@
+package plugin
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+)
+
+// A fetch response whose envelope does not fit the plugin's host buffer is
+// delivered in pieces: the host replies with status/headers plus a
+// body_handle, and the plugin pulls the body with {"body_handle","offset",
+// "limit"} requests, each chunk carried as base64 in the JSON envelope. This
+// file holds the transport-independent half so it can be unit tested without
+// a WASM runtime; wasm_backends.go supplies the real host call.
+
+const (
+	// fetchChunkSize is the raw bytes requested per chunk (~350 KiB once
+	// base64-encoded), small enough to fit a fresh 1 MiB buffer with room to
+	// spare.
+	fetchChunkSize = 256 * 1024
+	// minFetchChunkSize is where retries stop shrinking the chunk.
+	minFetchChunkSize = 4 * 1024
+	// maxFetchBodyBytes caps the body size a host may announce, so a bad
+	// body_size becomes an error instead of an allocation that traps the
+	// guest. It sits above the host's own 50 MiB per-response cap.
+	maxFetchBodyBytes = 64 << 20
+)
+
+// fetchEnvelope is the paca.fetch response: a [FetchResponse], plus the
+// chunking fields set when the body was parked host-side.
+type fetchEnvelope struct {
+	FetchResponse
+	BodyHandle string `json:"body_handle"`
+	BodySize   int    `json:"body_size"`
+}
+
+// fetchChunk is the response to a body_handle read.
+type fetchChunk struct {
+	Chunk    []byte `json:"chunk"` // base64 on the wire
+	BodySize int    `json:"body_size"`
+	Done     bool   `json:"done"`
+	Error    string `json:"error"`
+	Code     string `json:"code"`
+}
+
+// chunkReader performs one body_handle read against the host.
+type chunkReader func(handle string, offset, limit int) (*fetchChunk, error)
+
+// readBodyChunks drains a parked body of size bytes. If a chunk does not fit
+// the host buffer it halves the chunk size and retries, down to
+// minFetchChunkSize, so a plugin whose buffer is partly used still succeeds.
+func readBodyChunks(handle string, size int, read chunkReader) (string, error) {
+	if size < 0 || size > maxFetchBodyBytes {
+		return "", fmt.Errorf("plugin: fetch: %w: host announced a body of %d bytes (limit %d)", ErrResponseTooLarge, size, maxFetchBodyBytes)
+	}
+	var body strings.Builder
+	body.Grow(size)
+	limit := fetchChunkSize
+	for {
+		c, err := read(handle, body.Len(), limit)
+		if err != nil {
+			// A reply that did not fit, or no data at all (an exhausted
+			// buffer on a host that cannot say so), may succeed smaller.
+			if (errors.Is(err, ErrResponseTooLarge) || errors.Is(err, errHostNoData)) && limit > minFetchChunkSize {
+				limit /= 2
+				continue
+			}
+			return "", err
+		}
+		if body.Len()+len(c.Chunk) > size {
+			return "", fmt.Errorf("plugin: fetch: host sent more than the announced %d bytes", size)
+		}
+		body.Write(c.Chunk)
+		if c.Done {
+			if body.Len() != size {
+				return "", fmt.Errorf("plugin: fetch: body is %d bytes, host announced %d", body.Len(), size)
+			}
+			return body.String(), nil
+		}
+		if len(c.Chunk) == 0 {
+			return "", fmt.Errorf("plugin: fetch: host returned an empty chunk at offset %d of %d", body.Len(), size)
+		}
+	}
+}

fetch_chunks_test.go

@@ -0,0 +1,121 @@
+package plugin
+
+import (
+	"encoding/json"
+	"errors"
+	"strings"
+	"testing"
+)
+
+// fakeHost serves body in limit-capped chunks and can reject chunks above
+// maxFit with ErrResponseTooLarge, like a plugin whose buffer is partly full.
+type fakeHost struct {
+	body   string
+	maxFit int
+	calls  []int // limits requested
+}
+
+func (f *fakeHost) read(_ string, offset, limit int) (*fetchChunk, error) {
+	f.calls = append(f.calls, limit)
+	if f.maxFit > 0 && limit > f.maxFit {
+		return nil, hostResultError("fetch", "does not fit", hostCodeResponseTooLarge)
+	}
+	end := min(offset+limit, len(f.body))
+	return &fetchChunk{Chunk: []byte(f.body[offset:end]), BodySize: len(f.body), Done: end == len(f.body)}, nil
+}
+
+func TestReadBodyChunks_Reassembles(t *testing.T) {
+	body := strings.Repeat("0123456789", 100_000) // 1 MB, > several chunks
+	h := &fakeHost{body: body}
+	got, err := readBodyChunks("h", len(body), h.read)
+	if err != nil || got != body {
+		t.Fatalf("err=%v, equal=%v", err, got == body)
+	}
+	if len(h.calls) != 4 { // ceil(1_000_000 / 262_144)
+		t.Errorf("calls = %d, want 4", len(h.calls))
+	}
+}
+
+func TestReadBodyChunks_ShrinksOnTooLarge(t *testing.T) {
+	body := strings.Repeat("x", 100_000)
+	h := &fakeHost{body: body, maxFit: 20_000}
+	got, err := readBodyChunks("h", len(body), h.read)
+	if err != nil || got != body {
+		t.Fatalf("err=%v, equal=%v", err, got == body)
+	}
+	if h.calls[0] != fetchChunkSize || h.calls[len(h.calls)-1] > 20_000 {
+		t.Errorf("limits did not shrink as expected: %v", h.calls)
+	}
+}
+
+func TestReadBodyChunks_GivesUpAtMinimum(t *testing.T) {
+	h := &fakeHost{body: strings.Repeat("x", 100_000), maxFit: 100}
+	_, err := readBodyChunks("h", 100_000, h.read)
+	if !errors.Is(err, ErrResponseTooLarge) {
+		t.Fatalf("err = %v, want ErrResponseTooLarge", err)
+	}
+}
+
+func TestReadBodyChunks_Errors(t *testing.T) {
+	boom := errors.New("boom")
+	if _, err := readBodyChunks("h", 5, func(string, int, int) (*fetchChunk, error) { return nil, boom }); !errors.Is(err, boom) {
+		t.Errorf("host error not propagated: %v", err)
+	}
+	empty := func(string, int, int) (*fetchChunk, error) { return &fetchChunk{}, nil }
+	if _, err := readBodyChunks("h", 5, empty); err == nil || !strings.Contains(err.Error(), "empty chunk") {
+		t.Errorf("empty chunk must not loop forever: %v", err)
+	}
+	short := func(string, int, int) (*fetchChunk, error) { return &fetchChunk{Chunk: []byte("ab"), Done: true}, nil }
+	if _, err := readBodyChunks("h", 5, short); err == nil || !strings.Contains(err.Error(), "announced") {
+		t.Errorf("size mismatch must be reported: %v", err)
+	}
+}
+
+// The wire shape must match the host: chunk is base64, the meta envelope
+// flattens FetchResponse next to body_handle.
+func TestFetchWireShapes(t *testing.T) {
+	var c fetchChunk
+	if err := json.Unmarshal([]byte(`{"chunk":"aGVsbG8=","body_size":5,"done":true}`), &c); err != nil || string(c.Chunk) != "hello" || !c.Done {
+		t.Fatalf("chunk decode: %v %+v", err, c)
+	}
+	var e fetchEnvelope
+	if err := json.Unmarshal([]byte(`{"status":200,"headers":{"A":"b"},"body_handle":"h1","body_size":42}`), &e); err != nil {
+		t.Fatal(err)
+	}
+	if e.Status != 200 || e.Headers["A"] != "b" || e.BodyHandle != "h1" || e.BodySize != 42 {
+		t.Fatalf("envelope decode: %+v", e)
+	}
+}
+
+func TestReadBodyChunks_RejectsBadAnnouncedSize(t *testing.T) {
+	never := func(string, int, int) (*fetchChunk, error) { t.Fatal("must not read"); return nil, nil }
+	for _, size := range []int{-1, maxFetchBodyBytes + 1} {
+		if _, err := readBodyChunks("h", size, never); !errors.Is(err, ErrResponseTooLarge) {
+			t.Errorf("size %d: err = %v, want ErrResponseTooLarge", size, err)
+		}
+	}
+}
+
+func TestReadBodyChunks_RejectsOverrun(t *testing.T) {
+	over := func(string, int, int) (*fetchChunk, error) { return &fetchChunk{Chunk: []byte("abcdef")}, nil }
+	if _, err := readBodyChunks("h", 5, over); err == nil || !strings.Contains(err.Error(), "more than the announced") {
+		t.Errorf("overrun must be reported: %v", err)
+	}
+}
+
+func TestReadBodyChunks_ShrinksOnNoData(t *testing.T) {
+	body := strings.Repeat("x", 50_000)
+	var calls []int
+	read := func(_ string, offset, limit int) (*fetchChunk, error) {
+		calls = append(calls, limit)
+		if limit > 20_000 {
+			return nil, errEmptyHostResponse("fetch")
+		}
+		end := min(offset+limit, len(body))
+		return &fetchChunk{Chunk: []byte(body[offset:end]), Done: end == len(body)}, nil
+	}
+	got, err := readBodyChunks("h", len(body), read)
+	if err != nil || got != body {
+		t.Fatalf("err=%v, equal=%v, calls=%v", err, got == body, calls)
+	}
+}

host_limits.go

@@ -0,0 +1,44 @@
+package plugin
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+)
+
+// ErrResponseTooLarge is returned (wrapped) when the host produced a result
+// but it could not be delivered because it does not fit in the plugin's
+// fixed host-allocation buffer. It is a size limit, not a network failure:
+// retry with less data (e.g. a smaller page size), not with backoff.
+//
+//	if errors.Is(err, plugin.ErrResponseTooLarge) { ... }
+var ErrResponseTooLarge = errors.New("response too large")
+
+// hostCodeResponseTooLarge is the "code" value hosts put in an error result
+// that maps to ErrResponseTooLarge. Keep in sync with the host's
+// errCodeResponseTooLarge (services/api/internal/platform/plugin/runtime.go).
+const hostCodeResponseTooLarge = "response_too_large"
+
+// hostResultError builds the error for a host result envelope that carried
+// an error message (and optionally a machine-readable code).
+func hostResultError(op, msg, code string) error {
+	// Host messages already carry the operation prefix ("fetch: ..."); drop it
+	// rather than printing "plugin: fetch: fetch: ...".
+	msg = strings.TrimPrefix(msg, op+": ")
+	if code == hostCodeResponseTooLarge {
+		return fmt.Errorf("plugin: %s: %w: %s", op, ErrResponseTooLarge, msg)
+	}
+	return fmt.Errorf("plugin: %s: %s", op, msg)
+}
+
+// errHostNoData is wrapped by [errEmptyHostResponse] so internal callers can
+// tell "no data" apart from other failures.
+var errHostNoData = errors.New("host returned no data")
+
+// errEmptyHostResponse is returned when a host call delivered no data at all.
+// Newer hosts always send an explicit error envelope instead, so reaching
+// this means an older host, an exhausted buffer, or a host-side failure --
+// not necessarily an unreachable network.
+func errEmptyHostResponse(op string) error {
+	return fmt.Errorf("plugin: %s: %w (the result may not have fit in the plugin's host buffer, or the host call failed; this does not necessarily mean the network is unreachable)", op, errHostNoData)
+}

host_limits_test.go

@@ -0,0 +1,39 @@
+package plugin
+
+import (
+	"errors"
+	"strings"
+	"testing"
+)
+
+func TestHostResultError_ResponseTooLarge(t *testing.T) {
+	err := hostResultError("fetch", "fetch: response of 1238310 bytes does not fit", hostCodeResponseTooLarge)
+	if !errors.Is(err, ErrResponseTooLarge) {
+		t.Fatalf("errors.Is(err, ErrResponseTooLarge) = false for %v", err)
+	}
+	for _, want := range []string{"plugin: fetch:", "response too large", "1238310"} {
+		if !strings.Contains(err.Error(), want) {
+			t.Errorf("error %q missing %q", err, want)
+		}
+	}
+}
+
+func TestHostResultError_PlainError(t *testing.T) {
+	err := hostResultError("fetch", "fetch: domain not permitted by plugin manifest", "")
+	if errors.Is(err, ErrResponseTooLarge) {
+		t.Fatalf("plain host error must not match ErrResponseTooLarge: %v", err)
+	}
+	if got, want := err.Error(), "plugin: fetch: domain not permitted by plugin manifest"; got != want {
+		t.Errorf("got %q, want %q", got, want)
+	}
+}
+
+func TestErrEmptyHostResponse_NotMisleading(t *testing.T) {
+	msg := errEmptyHostResponse("fetch").Error()
+	if strings.Contains(msg, "empty response from host") {
+		t.Errorf("message should not use the old ambiguous wording: %q", msg)
+	}
+	if !strings.Contains(msg, "does not necessarily mean the network is unreachable") {
+		t.Errorf("message should disclaim a network failure: %q", msg)
+	}
+}

native_backends.go

@@ -71,6 +71,7 @@ type FetchResponse struct {
 	Body    string            `json:"body"`
 	Headers map[string]string `json:"headers"`
 	Error   string            `json:"error"`
+	Code    string            `json:"code,omitempty"`
 }
 
 // Fetch always returns errNotWASM outside a WASM module.