diff --git a/internal/plugin/host.go b/internal/plugin/host.go index 60a6679f..c14a8439 100644 --- a/internal/plugin/host.go +++ b/internal/plugin/host.go @@ -621,8 +621,8 @@ func Fetch(ctx context.Context, r FetchRequest) (*http.Response, error) { id, c := h.begin(true) params := struct { FetchRequest - Body string `json:"body,omitempty"` - }{r, base64.StdEncoding.EncodeToString(r.Body)} + Body []byte `json:"body,omitempty"` + }{r, r.Body} if err := h.send(map[string]any{"id": id, "method": "fetch", "params": params}); err != nil { h.forget(id) return nil, err diff --git a/internal/plugin/host_send_test.go b/internal/plugin/host_send_test.go new file mode 100644 index 00000000..152d6f5e --- /dev/null +++ b/internal/plugin/host_send_test.go @@ -0,0 +1,387 @@ +package plugin + +import ( + "bytes" + "context" + "encoding/base64" + "encoding/json" + "errors" + "io" + "os/exec" + "strconv" + "sync" + "testing" +) + +// This file is the wire contract of Fetch's request encoding: the body reaches +// the child as the base64 string encoding/json writes for a []byte, and the +// frame magpie writes is the very one the older EncodeToString params produced, +// byte for byte. Everything here goes through the production Fetch and its +// send, against a host this test builds itself; no test here starts Bun, and +// nothing here reaches into the host's stream bookkeeping. + +// wireIn is a host's stdin. It keeps what magpie wrote when asked to, and tells +// the answering goroutine that a frame came, so every fetch is answered (a +// frame left unanswered would park the production Fetch). +type wireIn struct { + keep bool + wrote chan struct{} // buffered 1: one frame, taken by the driver + mu sync.Mutex + lines [][]byte +} + +func newWireIn(keep bool) *wireIn { return &wireIn{keep: keep, wrote: make(chan struct{}, 1)} } + +func (c *wireIn) Write(p []byte) (int, error) { + if c.keep { + c.mu.Lock() + c.lines = append(c.lines, append([]byte(nil), p...)) + c.mu.Unlock() + } + // Only a fetch frame needs an answer. An abort magpie sends when the body + // is dropped is left alone: answering it would move the driver's ids on + // from the host's own, and the next fetch would be answered as another. + if bytes.HasPrefix(p, []byte(`{"id":`)) && bytes.Contains(p[:min(len(p), 64)], []byte(`"method":"fetch"`)) { + c.wrote <- struct{}{} // the driver is a goroutine of its own: it never holds a host lock + } + return len(p), nil +} + +func (c *wireIn) Close() error { return nil } + +// lastLine is the frame magpie wrote last, nil when none came. +func (c *wireIn) lastLine() []byte { + c.mu.Lock() + defer c.mu.Unlock() + if len(c.lines) == 0 { + return nil + } + return c.lines[len(c.lines)-1] +} + +// last is the frame magpie wrote last. +func (c *wireIn) last(t *testing.T) []byte { + t.Helper() + line := c.lastLine() + if line == nil { + t.Fatal("magpie wrote nothing to the child") + } + return line +} + +// fetchFrame is the fetch request magpie wrote, whatever else it wrote (an +// abort follows when a test ends a body by cancelling it). +func (c *wireIn) fetchFrame(t *testing.T) []byte { + t.Helper() + c.mu.Lock() + defer c.mu.Unlock() + for _, line := range c.lines { + if bytes.HasPrefix(line, []byte(`{"id":`)) && bytes.Contains(line[:min(len(line), 64)], []byte(`"method":"fetch"`)) { + return line + } + } + t.Fatal("magpie wrote no fetch request") + return nil +} + +// newWireHost is the host the tests and benchmarks drive: its stdout the test +// writes, its stdin kept, its read and dispatch the real ones. +func newWireHost(tb testing.TB, in *wireIn) (*host, *io.PipeWriter) { + tb.Helper() + pr, pw := io.Pipe() + h := &host{cmd: &exec.Cmd{}, in: in, calls: map[int64]*call{}, dead: make(chan struct{})} + go h.read(pr) + tb.Cleanup(func() { _ = pw.CloseWithError(io.EOF) }) + return h, pw +} + +// fetchHost makes h the host Fetch talks to and answers each fetch with only +// a head; callers cancel the context afterward. byFrame takes the id off the frame +// (tests, which keep them); otherwise the ids are the host's own, in order +// (benchmarks, which must not pay for parsing). +func fetchHost(tb testing.TB, h *host, pw *io.PipeWriter, in *wireIn, byFrame bool) { + tb.Helper() + // A plugin-list change another test made would have get() retire this host + // for a real one, so the change is absorbed here first. The flag that says + // a host is stale is taken rather than dropped, and put back when the + // helper is done: a restart another magpie asked for still happens after + // this test, and a newer signal is never overwritten with false. + checkList() + hostMu.Lock() + prev, prevStale := current, hostStale.Swap(false) + current = h + hostMu.Unlock() + write := func(v any) { + b, err := json.Marshal(v) + if err != nil { + panic(err) + } + _, _ = pw.Write(append(b, '\n')) + } + stop := make(chan struct{}) + go func() { + id := int64(0) + for { + select { + case <-stop: + return + case <-in.wrote: + } + if byFrame { + var m struct { + ID int64 `json:"id"` + } + line := in.lastLine() + if line == nil || json.Unmarshal(bytes.TrimSuffix(line, []byte("\n")), &m) != nil { + continue + } + id = m.ID + } else { + id++ + } + if id == 0 { + continue + } + // the head the child would have sent. The answer is never sent: it + // would race the head through Fetch's select, so a test ends the + // call by cancelling its context instead. + write(map[string]any{"id": id, "event": "head", "status": 200, "headers": map[string]string{"content-type": "text/event-stream"}}) + } + }() + tb.Cleanup(func() { + close(stop) + hostMu.Lock() + current = prev + if prevStale { + hostStale.Store(true) + } + hostMu.Unlock() + }) +} + +// fetchOnce sends one request through the production Fetch and gives back the +// frame magpie wrote for it. +func fetchOnce(t *testing.T, r FetchRequest) []byte { + t.Helper() + in := newWireIn(true) + h, pw := newWireHost(t, in) + fetchHost(t, h, pw, in, true) + ctx, cancel := context.WithCancel(context.Background()) + res, err := Fetch(ctx, r) + if err != nil { + cancel() + t.Fatalf("Fetch: %v", err) + } + line := in.fetchFrame(t) // read before ending the call: the abort follows it + if res.StatusCode != 200 { + cancel() + t.Fatalf("status = %d", res.StatusCode) + } + cancel() // the host sent no chunks: the call ends by cancelling, not by an answer + if b, err := io.ReadAll(res.Body); err != nil && !errors.Is(err, context.Canceled) { + t.Fatalf("the body = %q, %v", b, err) + } else if len(b) != 0 { + t.Fatalf("the host sent no chunks, so the body = %q", b) + } + res.Body.Close() + return line +} + +// legacyFetchParams is the anonymous params struct Fetch built before the body +// went to encoding/json as a []byte: the same tags, the body as the base64 +// string EncodeToString made of it. +func legacyFetchParams(r FetchRequest) any { + return struct { + FetchRequest + Body string `json:"body,omitempty"` + }{r, base64.StdEncoding.EncodeToString(r.Body)} +} + +// legacyFrame is the frame the older Fetch would have written for r, id and all. +func legacyFrame(t *testing.T, id int64, r FetchRequest) []byte { + t.Helper() + b, err := json.Marshal(map[string]any{"id": id, "method": "fetch", "params": legacyFetchParams(r)}) + if err != nil { + t.Fatalf("marshal: %v", err) + } + return append(b, '\n') +} + +// frameOf is a written frame, taken apart. +type frameOf struct { + ID int64 `json:"id"` + Method string `json:"method"` + Params map[string]json.RawMessage `json:"params"` +} + +func parseFrame(t *testing.T, line []byte) frameOf { + t.Helper() + var m frameOf + if err := json.Unmarshal(bytes.TrimSuffix(line, []byte("\n")), &m); err != nil { + t.Fatalf("the frame is not one JSON object: %v (%q)", err, clip(line)) + } + return m +} + +// clip shortens a frame for a failure message: a 1MiB body must not fill one. +func clip(b []byte) string { + const n = 200 + if len(b) <= n { + return string(b) + } + return string(b[:n]) + "…(" + strconv.Itoa(len(b)) + " bytes)" +} + +// TestFetchHostKeepsTheRestartSignal: installing the test's host takes the flag +// that says the plugin list changed (get() would otherwise retire it for a real +// host), so the helper has to put that flag back when it is done — a restart +// another magpie asked for must not be dropped by a test. +func TestFetchHostKeepsTheRestartSignal(t *testing.T) { + prev := hostStale.Load() + t.Cleanup(func() { hostStale.Store(prev) }) // these tests run one at a time + hostMu.Lock() + before := current + hostMu.Unlock() + hostStale.Store(true) // what another magpie's change leaves behind + + t.Run("a helper used and cleaned up", func(t *testing.T) { + in := newWireIn(false) + h, pw := newWireHost(t, in) + fetchHost(t, h, pw, in, false) + if hostStale.Load() { + t.Fatal("the test's host is installed with the stale flag still set: get() would retire it") + } + }) + + if !hostStale.Load() { + t.Fatal("the restart signal did not survive the helper's cleanup") + } + hostMu.Lock() + after := current + hostMu.Unlock() + if after != before { + t.Fatalf("current = %p, want the host from before the helper (%p)", after, before) + } +} + +func TestFetchBodyWireIsUnchanged(t *testing.T) { + htmlUnicode := "a&\"c\" — 你好