plugin: reuse config parsing and avoid request body string copies (#696)
This commit is contained in:
1 parent
6c46e5561c
commit
d62fecd468
5 files changed
+782
-11
No files matched your search
@@ -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
|
||||
|
||||
@@ -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<b>&\"c\" — 你好 <script>"
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
body []byte
|
||||
}{
|
||||
{"json text", []byte(`{"model":"fake-1","messages":[{"role":"user","content":"héllo — 你好 <b>&</b>"}]}`)},
|
||||
{"binary", []byte{0x00, 0x01, 0xfb, 0xff, 0x80, 0x7f}},
|
||||
{"one byte", []byte("a")}, // base64 "=" padding
|
||||
{"two bytes", []byte("ab")}, // "=="
|
||||
{"three bytes", []byte("abc")}, // no padding
|
||||
{"nil body", nil},
|
||||
{"empty body", []byte{}},
|
||||
{"1MiB", bytes.Repeat([]byte("0123456789abcdef"), 1<<16)},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
r := FetchRequest{
|
||||
Provider: "fakeco", Account: "acct-1", Model: "fake-1", NPM: "@ai-sdk/openai-compatible",
|
||||
URL: "https://vendor.example/v1/chat/completions?model=a&x=<b>&y=%E4%BD%A0%E5%A5%BD",
|
||||
Method: "POST",
|
||||
Headers: map[string]string{"content-type": "application/json", "x-note": htmlUnicode},
|
||||
Body: tc.body, Session: "s-1", Proxy: "direct",
|
||||
}
|
||||
line := fetchOnce(t, r)
|
||||
|
||||
// one frame, newline-terminated, nothing raw inside it
|
||||
if !bytes.HasSuffix(line, []byte("\n")) || bytes.Count(line, []byte("\n")) != 1 {
|
||||
t.Fatalf("framing: %q", clip(line))
|
||||
}
|
||||
got := parseFrame(t, line)
|
||||
if got.Method != "fetch" || got.ID == 0 {
|
||||
t.Fatalf("frame = %+v", got)
|
||||
}
|
||||
|
||||
// the older params build the same frame, byte for byte (the
|
||||
// metadata above is there to keep escaping and ordering pinned)
|
||||
want := legacyFrame(t, got.ID, r)
|
||||
if !bytes.Equal(line, want) {
|
||||
t.Fatalf("the wire changed:\n got %s\nwant %s", clip(line), clip(want))
|
||||
}
|
||||
|
||||
// the body's field presence and its bytes, off the wire itself
|
||||
wantParams := parseFrame(t, want).Params
|
||||
raw, present := got.Params["body"]
|
||||
_, wantPresent := wantParams["body"]
|
||||
if present != wantPresent {
|
||||
t.Fatalf("the body field is there %v, the older frame has it %v", present, wantPresent)
|
||||
}
|
||||
if !present {
|
||||
return
|
||||
}
|
||||
var b64 string
|
||||
if err := json.Unmarshal(raw, &b64); err != nil {
|
||||
t.Fatalf("the body field is not a base64 string: %v", err)
|
||||
}
|
||||
back, err := base64.StdEncoding.DecodeString(b64)
|
||||
if err != nil {
|
||||
t.Fatalf("the body does not decode: %v", err)
|
||||
}
|
||||
if !bytes.Equal(back, tc.body) {
|
||||
t.Fatalf("the body came back %d bytes, want %d", len(back), len(tc.body))
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// benchBodies are the payloads the benchmark sends, built once, outside the
|
||||
// timer: what is timed is the encoding and the send, not the making of a body.
|
||||
func benchBodies() []struct {
|
||||
name string
|
||||
body []byte
|
||||
} {
|
||||
return []struct {
|
||||
name string
|
||||
body []byte
|
||||
}{
|
||||
{"1MiB", bytes.Repeat([]byte("0123456789abcdef"), 1<<16)},
|
||||
{"8MiB", bytes.Repeat([]byte("0123456789abcdef"), 1<<19)},
|
||||
{"16MiB", bytes.Repeat([]byte("0123456789abcdef"), 1<<20)},
|
||||
}
|
||||
}
|
||||
|
||||
func benchRequest(body []byte) FetchRequest {
|
||||
return FetchRequest{
|
||||
Provider: "fakeco", Account: "acct-1", Model: "fake-1", NPM: "@ai-sdk/openai-compatible",
|
||||
URL: "https://vendor.example/chat/completions",
|
||||
Method: "POST",
|
||||
Headers: map[string]string{"content-type": "application/json"},
|
||||
Body: body, Session: "s-1", Proxy: "direct",
|
||||
}
|
||||
}
|
||||
|
||||
// BenchmarkFetchBodyProd is the production path: Fetch builds its params with
|
||||
// the body as a []byte and sends them. The pipe write, head dispatch and
|
||||
// cancellation cleanup are part of every iteration. The paired comparison
|
||||
// with the older encoding is this same benchmark run twice, overlaying host.go
|
||||
// with the old Fetch body representation — there is no mirrored copy of it here
|
||||
// to drift from the production path.
|
||||
func BenchmarkFetchBodyProd(b *testing.B) {
|
||||
for _, tc := range benchBodies() {
|
||||
b.Run(tc.name, func(b *testing.B) {
|
||||
in := newWireIn(false)
|
||||
h, pw := newWireHost(b, in)
|
||||
fetchHost(b, h, pw, in, false)
|
||||
r := benchRequest(tc.body)
|
||||
b.SetBytes(int64(len(tc.body)))
|
||||
b.ReportAllocs()
|
||||
for b.Loop() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
res, err := Fetch(ctx, r)
|
||||
if err != nil {
|
||||
cancel()
|
||||
b.Fatal(err)
|
||||
}
|
||||
cancel() // ends the call: the harness sends a head and nothing else
|
||||
_, _ = io.Copy(io.Discard, res.Body)
|
||||
res.Body.Close()
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -347,7 +347,7 @@ func keepUnloaded(ps, last []Provider) []Provider {
|
||||
told[p.Spec] = true
|
||||
}
|
||||
installed := map[string]bool{}
|
||||
for _, e := range Load().Plugins {
|
||||
for _, e := range list().Plugins {
|
||||
installed[e.Spec] = true
|
||||
}
|
||||
for _, p := range last {
|
||||
@@ -407,6 +407,7 @@ func Settle() {
|
||||
// provider's sign-in is read afresh from plugin-auth.json.
|
||||
func Cached() []Provider {
|
||||
checkList()
|
||||
l := list() // one read of plugins.json for the whole call
|
||||
provMu.Lock()
|
||||
ps := provCache
|
||||
good := provGood
|
||||
@@ -416,13 +417,13 @@ func Cached() []Provider {
|
||||
_ = json.Unmarshal(b, &ps)
|
||||
}
|
||||
}
|
||||
if len(Load().Plugins) == 0 {
|
||||
if len(l.Plugins) == 0 {
|
||||
return nil
|
||||
}
|
||||
auth := readAuth()
|
||||
out := make([]Provider, 0, len(ps))
|
||||
on := map[string]bool{}
|
||||
for _, e := range Load().Plugins {
|
||||
for _, e := range l.Plugins {
|
||||
if !e.Off {
|
||||
on[e.Spec] = true
|
||||
}
|
||||
@@ -448,7 +449,7 @@ func Cached() []Provider {
|
||||
}
|
||||
out = append(out, p)
|
||||
}
|
||||
if !good && len(Load().Plugins) > 0 {
|
||||
if !good && len(l.Plugins) > 0 {
|
||||
// refreshed in the background: a sign-in or the plugins changed.
|
||||
// Whether one is already in flight is read and set under one lock,
|
||||
// so two callers asking at once start one refresh, not two.
|
||||
|
||||
@@ -17,6 +17,7 @@
|
||||
package plugin
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
@@ -96,15 +97,88 @@ func lockAuth() func() {
|
||||
|
||||
func listPath() string { return filepath.Join(settings.Dir(), "plugins.json") }
|
||||
|
||||
// Load reads plugins.json.
|
||||
func Load() List {
|
||||
var l List
|
||||
if b, err := steady.ReadFile(listPath()); err == nil {
|
||||
_ = json.Unmarshal(b, &l)
|
||||
// pluginsList is plugins.json as last read, with the bytes it was parsed
|
||||
// from. The file is read every time: an edit that keeps the file's size and
|
||||
// time (a swap of two accounts of one length, a replaced file whose time is
|
||||
// put back) would be missed by a stamp, and only the parse is reused.
|
||||
var pluginsList struct {
|
||||
sync.Mutex
|
||||
path string
|
||||
raw []byte
|
||||
list List
|
||||
}
|
||||
|
||||
// list is plugins.json, read now. What it gives is shared and must not be
|
||||
// changed; Load gives the mutable copy.
|
||||
func list() List {
|
||||
path := listPath()
|
||||
b, err := steady.ReadFile(path)
|
||||
if err != nil {
|
||||
// Do not cache read failures; a later call will read the file again.
|
||||
return List{}
|
||||
}
|
||||
pluginsList.Lock()
|
||||
defer pluginsList.Unlock()
|
||||
if pluginsList.path == path && bytes.Equal(pluginsList.raw, b) {
|
||||
return pluginsList.list
|
||||
}
|
||||
var l List
|
||||
// Preserve partially decoded values on type errors, as Load has always done.
|
||||
_ = json.Unmarshal(b, &l)
|
||||
pluginsList.path, pluginsList.raw, pluginsList.list = path, b, l
|
||||
return l
|
||||
}
|
||||
|
||||
// Load reads plugins.json. What it gives is the caller's to change at every
|
||||
// level: the mutators (Add, Remove, SetOff, SetConfig) each get their own
|
||||
// copy, so nothing they change reaches what list keeps.
|
||||
func Load() List {
|
||||
l := list()
|
||||
return List{Plugins: clonePlugins(l.Plugins), Config: cloneMap(l.Config)}
|
||||
}
|
||||
|
||||
// clonePlugins copies the entries, their options with them.
|
||||
func clonePlugins(es []Entry) []Entry {
|
||||
if es == nil {
|
||||
return nil
|
||||
}
|
||||
out := make([]Entry, len(es))
|
||||
for i, e := range es {
|
||||
out[i] = e
|
||||
out[i].Options = cloneMap(e.Options)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// cloneMap copies a JSON object and everything under it.
|
||||
func cloneMap(m map[string]any) map[string]any {
|
||||
if m == nil {
|
||||
return nil
|
||||
}
|
||||
out := make(map[string]any, len(m))
|
||||
for k, v := range m {
|
||||
out[k] = cloneJSON(v)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// cloneJSON copies a JSON value: Unmarshal into any gives only objects,
|
||||
// arrays, strings, numbers, bools and nil, so this reaches every part of one.
|
||||
func cloneJSON(v any) any {
|
||||
switch x := v.(type) {
|
||||
case map[string]any:
|
||||
return cloneMap(x)
|
||||
case []any:
|
||||
out := make([]any, len(x))
|
||||
for i, e := range x {
|
||||
out[i] = cloneJSON(e)
|
||||
}
|
||||
return out
|
||||
default:
|
||||
return v
|
||||
}
|
||||
}
|
||||
|
||||
// writeWhole writes b to p by a rename, so a magpie or the host reading
|
||||
// p meanwhile reads the old file or the new one, never one half-written;
|
||||
// read back with steady.ReadFile, which waits out the rename on Windows.
|
||||
|
||||
@@ -0,0 +1,309 @@
|
||||
package plugin
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/yetone/magpie/internal/settings"
|
||||
)
|
||||
|
||||
// storeSandbox gives the test its own magpie folders, without the Bun the
|
||||
// other tests need: these only read and write plugins.json.
|
||||
func storeSandbox(t testing.TB) string {
|
||||
t.Helper()
|
||||
dir := t.TempDir()
|
||||
t.Setenv("HOME", dir)
|
||||
t.Setenv("XDG_CONFIG_HOME", filepath.Join(dir, "config"))
|
||||
t.Setenv("XDG_CACHE_HOME", filepath.Join(dir, "cache"))
|
||||
if err := os.MkdirAll(settings.Dir(), 0o700); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return dir
|
||||
}
|
||||
|
||||
// writeList puts plugins.json where magpie keeps it.
|
||||
func writeList(t testing.TB, body string) {
|
||||
t.Helper()
|
||||
if err := os.WriteFile(listPath(), []byte(body), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoadMissingIsEmptyNotGrown(t *testing.T) {
|
||||
storeSandbox(t)
|
||||
l := Load()
|
||||
if l.Plugins != nil || l.Config != nil {
|
||||
t.Fatalf("a missing plugins.json gave %+v, want the zero List", l)
|
||||
}
|
||||
}
|
||||
|
||||
// TestLoadKeepsWhatAPartialFileGave: what a file that doesn't parse whole
|
||||
// gives is what Load gives, as it always was — the error is no more read
|
||||
// than it was, so a half-written file can't empty a list a magpie is using.
|
||||
func TestLoadKeepsWhatAPartialFileGave(t *testing.T) {
|
||||
for _, tc := range []struct{ name, body string }{
|
||||
{"truncated", `{"plugins":[{"spec":"a"},{"spec":"b"`},
|
||||
{"a field of the wrong type", `{"plugins":[{"spec":"a"},{"spec":"b","off":"yes"}],"config":{"k":1}}`},
|
||||
{"empty", `{"plugins":[]}`},
|
||||
{"whole", `{"plugins":[{"spec":"a","options":{"k":1}}],"config":{"p":{"q":["x"]}}}`},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
storeSandbox(t)
|
||||
writeList(t, tc.body)
|
||||
// what a plain parse of those bytes gives is what Load gives:
|
||||
// the error was never read, and a partly filled list stayed
|
||||
var want List
|
||||
_ = json.Unmarshal([]byte(tc.body), &want)
|
||||
if got := Load(); !reflect.DeepEqual(got, want) {
|
||||
t.Fatalf("Load = %+v, want the plain parse %+v", got, want)
|
||||
}
|
||||
})
|
||||
}
|
||||
t.Run("a plugin edited after a wrong type isn't saved away", func(t *testing.T) {
|
||||
storeSandbox(t)
|
||||
writeList(t, `{"plugins":[{"spec":"a"},{"spec":"b","off":"yes"}],"config":{"k":1}}`)
|
||||
if l := Load(); len(l.Plugins) != 2 || l.Plugins[0].Spec != "a" || l.Plugins[1].Spec != "b" || l.Plugins[1].Off {
|
||||
t.Fatalf("a wrong type gave %+v, want both plugins", l.Plugins)
|
||||
}
|
||||
if err := SetOff("a", true); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if l := Load(); len(l.Plugins) != 2 || !l.Plugins[0].Off || l.Plugins[1].Spec != "b" || l.Config["k"] != float64(1) {
|
||||
t.Fatalf("after turning one off: %+v", l)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// TestLoadCopyIsIndependent: what Load gives may be changed anywhere in it,
|
||||
// and what magpie keeps is not.
|
||||
func TestLoadCopyIsIndependent(t *testing.T) {
|
||||
storeSandbox(t)
|
||||
writeList(t, `{"plugins":[{"spec":"a","options":{"k":{"n":[1]}}}],"config":{"p":{"q":["x"]}}}`)
|
||||
l := Load()
|
||||
l.Plugins[0].Spec = "changed"
|
||||
l.Plugins[0].Off = true
|
||||
l.Plugins[0].Options["k"].(map[string]any)["n"].([]any)[0] = 2
|
||||
l.Plugins = append(l.Plugins, Entry{Spec: "added"})
|
||||
l.Config["p"].(map[string]any)["q"].([]any)[0] = "changed"
|
||||
l.Config["new"] = 1
|
||||
|
||||
again := Load()
|
||||
if len(again.Plugins) != 1 || again.Plugins[0].Spec != "a" || again.Plugins[0].Off {
|
||||
t.Fatalf("the entries changed with a copy: %+v", again.Plugins)
|
||||
}
|
||||
if n := again.Plugins[0].Options["k"].(map[string]any)["n"].([]any)[0]; n != float64(1) {
|
||||
t.Fatalf("a nested option changed with a copy: %v", n)
|
||||
}
|
||||
if q := again.Config["p"].(map[string]any)["q"].([]any)[0]; q != "x" {
|
||||
t.Fatalf("a nested config value changed with a copy: %v", q)
|
||||
}
|
||||
if _, ok := again.Config["new"]; ok {
|
||||
t.Fatal("a new config key reached what magpie keeps")
|
||||
}
|
||||
}
|
||||
|
||||
// TestListReusesTheParse: while the bytes are the same the parse is reused.
|
||||
func TestListReusesTheParse(t *testing.T) {
|
||||
storeSandbox(t)
|
||||
writeList(t, `{"plugins":[{"spec":"a"},{"spec":"b"}]}`)
|
||||
a := list()
|
||||
b := list()
|
||||
if len(a.Plugins) == 0 || &a.Plugins[0] != &b.Plugins[0] {
|
||||
t.Fatal("the same bytes were parsed again")
|
||||
}
|
||||
}
|
||||
|
||||
// TestLoadSeesAnEditThatKeepsSizeAndTime: an old file edited in place, or
|
||||
// replaced by a rename, with its size and time put back, is read as it is
|
||||
// now — a stamp would serve the parse before it.
|
||||
func TestLoadSeesAnEditThatKeepsSizeAndTime(t *testing.T) {
|
||||
old := time.Now().Add(-time.Hour)
|
||||
edit := func(t *testing.T, was, now string) {
|
||||
t.Helper()
|
||||
writeList(t, was)
|
||||
if err := os.Chtimes(listPath(), old, old); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if l := Load(); len(l.Plugins) != 1 || l.Plugins[0].Spec != specOf(was) {
|
||||
t.Fatalf("first read: %+v", l.Plugins)
|
||||
}
|
||||
writeList(t, now)
|
||||
if err := os.Chtimes(listPath(), old, old); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if l := Load(); len(l.Plugins) != 1 || l.Plugins[0].Spec != specOf(now) {
|
||||
t.Fatalf("an edit of one size kept the parse before it: %+v", l.Plugins)
|
||||
}
|
||||
}
|
||||
t.Run("in place", func(t *testing.T) {
|
||||
storeSandbox(t)
|
||||
edit(t, `{"plugins":[{"spec":"aaa"}]}`, `{"plugins":[{"spec":"bbb"}]}`)
|
||||
})
|
||||
t.Run("replaced by a rename", func(t *testing.T) {
|
||||
storeSandbox(t)
|
||||
writeList(t, `{"plugins":[{"spec":"aaa"}]}`)
|
||||
if err := os.Chtimes(listPath(), old, old); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if l := Load(); len(l.Plugins) != 1 || l.Plugins[0].Spec != "aaa" {
|
||||
t.Fatalf("first read: %+v", l.Plugins)
|
||||
}
|
||||
tmp := filepath.Join(filepath.Dir(listPath()), "plugins.json.tmp")
|
||||
if err := os.WriteFile(tmp, []byte(`{"plugins":[{"spec":"bbb"}]}`), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.Chtimes(tmp, old, old); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.Rename(tmp, listPath()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if l := Load(); len(l.Plugins) != 1 || l.Plugins[0].Spec != "bbb" {
|
||||
t.Fatalf("a replaced file kept the parse before it: %+v", l.Plugins)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// specOf is the one plugin spec a test's list holds.
|
||||
func specOf(body string) string {
|
||||
var l List
|
||||
_ = json.Unmarshal([]byte(body), &l)
|
||||
if len(l.Plugins) == 0 {
|
||||
return ""
|
||||
}
|
||||
return l.Plugins[0].Spec
|
||||
}
|
||||
|
||||
// TestLoadDeleteRecreateAndReadError: gone, unreadable and back again.
|
||||
func TestLoadDeleteRecreateAndReadError(t *testing.T) {
|
||||
storeSandbox(t)
|
||||
writeList(t, `{"plugins":[{"spec":"a"}]}`)
|
||||
if l := Load(); len(l.Plugins) != 1 {
|
||||
t.Fatalf("first read: %+v", l.Plugins)
|
||||
}
|
||||
if err := os.Remove(listPath()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if l := Load(); l.Plugins != nil {
|
||||
t.Fatalf("a deleted file gave %+v, want the zero List", l.Plugins)
|
||||
}
|
||||
// a folder where the file goes: reading it fails
|
||||
if err := os.Mkdir(listPath(), 0o700); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if l := Load(); l.Plugins != nil {
|
||||
t.Fatalf("an unreadable file gave %+v, want the zero List", l.Plugins)
|
||||
}
|
||||
if err := os.Remove(listPath()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
writeList(t, `{"plugins":[{"spec":"back"}]}`)
|
||||
if l := Load(); len(l.Plugins) != 1 || l.Plugins[0].Spec != "back" {
|
||||
t.Fatalf("after it came back: %+v", l.Plugins)
|
||||
}
|
||||
}
|
||||
|
||||
// TestLoadPerConfigDir: what one magpie folder keeps is not another's.
|
||||
func TestLoadPerConfigDir(t *testing.T) {
|
||||
dir := storeSandbox(t)
|
||||
writeList(t, `{"plugins":[{"spec":"one"}]}`)
|
||||
if l := Load(); l.Plugins[0].Spec != "one" {
|
||||
t.Fatalf("first dir: %+v", l.Plugins)
|
||||
}
|
||||
other := filepath.Join(dir, "elsewhere")
|
||||
t.Setenv("XDG_CONFIG_HOME", other)
|
||||
if err := os.MkdirAll(settings.Dir(), 0o700); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
writeList(t, `{"plugins":[{"spec":"two"}]}`)
|
||||
if l := Load(); len(l.Plugins) != 1 || l.Plugins[0].Spec != "two" {
|
||||
t.Fatalf("second dir: %+v", l.Plugins)
|
||||
}
|
||||
t.Setenv("XDG_CONFIG_HOME", filepath.Join(dir, "config"))
|
||||
if l := Load(); len(l.Plugins) != 1 || l.Plugins[0].Spec != "one" {
|
||||
t.Fatalf("back to the first dir: %+v", l.Plugins)
|
||||
}
|
||||
}
|
||||
|
||||
// TestLoadConcurrentWithSavesAndEdits: reading the list while another
|
||||
// magpie's writes land (each of them a rename) never gives half a file, and
|
||||
// add, off and remove work under it.
|
||||
func TestLoadConcurrentWithSavesAndEdits(t *testing.T) {
|
||||
dir := storeSandbox(t)
|
||||
writeList(t, `{"plugins":[{"spec":"p0"}]}`)
|
||||
plug := filepath.Join(dir, "plug.mjs")
|
||||
if err := os.WriteFile(plug, []byte("// a plugin\n"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
stop := make(chan struct{})
|
||||
var wg sync.WaitGroup
|
||||
for range 4 {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for {
|
||||
select {
|
||||
case <-stop:
|
||||
return
|
||||
default:
|
||||
}
|
||||
plugins := Load().Plugins
|
||||
if len(plugins) < 1 || len(plugins) > 2 {
|
||||
t.Errorf("a read gave %d plugins", len(plugins))
|
||||
return
|
||||
}
|
||||
for _, e := range plugins {
|
||||
if e.Spec == "" {
|
||||
t.Errorf("a read gave an entry with no spec")
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer close(stop)
|
||||
for i := range 60 {
|
||||
_ = save(List{Plugins: []Entry{{Spec: fmt.Sprintf("p%d", i%2)}}})
|
||||
if _, err := Add(context.Background(), plug); err != nil {
|
||||
t.Errorf("add: %v", err)
|
||||
return
|
||||
}
|
||||
if err := SetOff(Name(plug), true); err != nil {
|
||||
t.Errorf("off: %v", err)
|
||||
return
|
||||
}
|
||||
if err := Remove(context.Background(), Name(plug)); err != nil {
|
||||
t.Errorf("remove: %v", err)
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
wg.Wait()
|
||||
if l := Load(); len(l.Plugins) != 1 || !strings.HasPrefix(l.Plugins[0].Spec, "p") {
|
||||
t.Fatalf("after the writers: %+v", l.Plugins)
|
||||
}
|
||||
}
|
||||
|
||||
// BenchmarkCachedWarm is Cached with the providers already good, so nothing
|
||||
// is refreshed in the background: what is measured is the per-call read.
|
||||
func BenchmarkCachedWarm(b *testing.B) {
|
||||
storeSandbox(b)
|
||||
writeList(b, `{"plugins":[{"spec":"@magpie-community/opencode-zed-auth@latest","options":{"a":1}},{"spec":"@magpie-community/opencode-qoder-auth@latest","off":true}],"config":{"provider":{"zed":{"options":{"projectId":"p"}}}}}`)
|
||||
UseCached([]Provider{{ID: "zed", Spec: "@magpie-community/opencode-zed-auth@latest", Name: "Zed", SignedIn: true}})
|
||||
b.ReportAllocs()
|
||||
for b.Loop() {
|
||||
Cached()
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user