From 4cc32dbcb0b22ea0d2a3e8faba55c567a26a9eb0 Mon Sep 17 00:00:00 2001 From: yetone Date: Tue, 29 Sep 2026 02:22:35 +0800 Subject: [PATCH] =?UTF-8?q?routing:=20every=20request's=20route=20is=20kep?= =?UTF-8?q?t=20on=20disk,=20for=20a=20day=20long=20gone=20to=20be=20looked?= =?UTF-8?q?=20at=20and=20replayed,=20with=20retention=20(=E5=8E=86?= =?UTF-8?q?=E5=8F=B2=E4=B8=AD=E7=9A=84=E8=B7=AF=E7=94=B1=E8=AE=B0=E5=BD=95?= =?UTF-8?q?=E6=98=AF=E4=B8=8D=E6=98=AF=E9=83=BD=E5=BA=94=E8=AF=A5=E4=BF=9D?= =?UTF-8?q?=E5=AD=98=E4=B8=8B=E6=9D=A5=EF=BC=8C=E6=96=B9=E4=BE=BF=E7=94=A8?= =?UTF-8?q?=E6=88=B7=E5=8E=BB=E5=9B=9E=E6=94=BE=EF=BC=9F=E2=80=A6=E8=BF=99?= =?UTF-8?q?=E7=A7=8D=E5=8E=86=E5=8F=B2=E8=AE=B0=E5=BD=95=E4=BC=9A=E7=89=B9?= =?UTF-8?q?=E5=88=AB=E5=A4=A7=EF=BC=8C=E5=BA=94=E8=AF=A5=E6=9C=89=E8=87=AA?= =?UTF-8?q?=E5=8A=A8=E7=9A=84retention)=20=E2=80=94=20the=20trace=20holds?= =?UTF-8?q?=20only=20the=20last=2060=20routes,=20and=20only=20since=20the?= =?UTF-8?q?=20gateway=20started.=20Now=20each=20route,=20once=20done,=20is?= =?UTF-8?q?=20appended=20to=20~/.config/magpie/routing/YYYY-MM-DD.jsonl;?= =?UTF-8?q?=20a=20day=20that=20is=20over=20is=20gzipped,=20days=20older=20?= =?UTF-8?q?than=2030=20are=20dropped,=20then=20the=20oldest=20until=20all?= =?UTF-8?q?=20of=20it=20fits=20200=20MB=20(today=20is=20never=20dropped),?= =?UTF-8?q?=20checked=20at=20most=20hourly.=20Route=20ids=20start=20at=20t?= =?UTF-8?q?he=20time=20the=20gateway=20started,=20so=20they=20don't=20repe?= =?UTF-8?q?at=20across=20runs.=20/api/gateway/history=3Fday=3D=20gives=20t?= =?UTF-8?q?he=20days=20kept=20with=20their=20counts=20and=20a=20day's=20ro?= =?UTF-8?q?utes=20(its=20latest=202000).=20The=20Requests=20list=20has=20a?= =?UTF-8?q?=20chip=20per=20day=20(=E5=AE=9E=E6=97=B6,=20=E4=BB=8A=E5=A4=A9?= =?UTF-8?q?=203,=20=E6=98=A8=E5=A4=A9,=20=E2=80=A6):=20a=20day's=20request?= =?UTF-8?q?s=20can=20be=20picked,=20looked=20at,=20replayed=20all,=20or=20?= =?UTF-8?q?replayed=20from=20one=20on=20(=E4=BB=8E=E8=BF=99=E9=87=8C?= =?UTF-8?q?=E9=87=8D=E6=94=BE),=20with=20the=20real-time=20clock=20as=20be?= =?UTF-8?q?fore.=20Checked=20in=20a=20sandbox=20HOME=20against=20real=20De?= =?UTF-8?q?epSeek:=20three=20requests=20written=20to=20the=20day's=20file,?= =?UTF-8?q?=20the=20gateway=20restarted,=20the=20list=20showed=20=E4=BB=8A?= =?UTF-8?q?=E5=A4=A9=203=20with=20them,=20=E4=BB=8E=E8=BF=99=E9=87=8C?= =?UTF-8?q?=E9=87=8D=E6=94=BE=20played=20them=20at=2002:19:46=20=C2=B7=20?= =?UTF-8?q?=E7=AC=AC=201/3,=20stop=20and=20=E5=AE=9E=E6=97=B6=20came=20bac?= =?UTF-8?q?k=20to=20live?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/gateway/history.go | 207 +++++++++++++++++++++++++++++++ internal/gateway/history_test.go | 52 ++++++++ internal/gateway/trace.go | 10 ++ internal/gui/assets/i18n.js | 5 + internal/gui/assets/routing.css | 8 ++ internal/gui/assets/routing.js | 71 +++++++++-- internal/gui/trace.go | 6 + 7 files changed, 351 insertions(+), 8 deletions(-) create mode 100644 internal/gateway/history.go create mode 100644 internal/gateway/history_test.go diff --git a/internal/gateway/history.go b/internal/gateway/history.go new file mode 100644 index 00000000..251450ce --- /dev/null +++ b/internal/gateway/history.go @@ -0,0 +1,207 @@ +package gateway + +// The routing history: every request's route, once it is done, kept on +// disk so the Routing view can show and replay a day long gone — the trace +// holds only the last few, and only since the gateway started. A day is a +// file of its own, one route a line; a day that is over is gzipped, the +// routes being much alike. Kept historyDays days and historyBytes in all, +// the oldest days going first. + +import ( + "bufio" + "bytes" + "compress/gzip" + "encoding/json" + "io" + "os" + "path/filepath" + "sort" + "strings" + "sync" + "time" + + "github.com/yetone/magpie/internal/provider" +) + +const ( + historyDays = 30 + historyBytes = 200 << 20 + historyMax = 2000 // the routes of one day served at most: its latest +) + +// HistoryDir is where the days are kept: ~/.config/magpie/routing. +func HistoryDir() string { return filepath.Join(filepath.Dir(provider.Path()), "routing") } + +var history struct { + mu sync.Mutex + pruned time.Time + counts map[string]counted // a day's file → its requests, as of its size +} + +type counted struct { + size int64 + n int +} + +const dayForm = "2006-01-02" + +// saveRoute adds a done route to its day. Errors are swallowed: keeping +// the history must never break a call. +func saveRoute(r Route) { + r.Seq = 0 + b, err := json.Marshal(r) + if err != nil { + return + } + history.mu.Lock() + defer history.mu.Unlock() + dir := HistoryDir() + if os.MkdirAll(dir, 0o755) != nil { + return + } + f, err := os.OpenFile(filepath.Join(dir, r.Time.Local().Format(dayForm)+".jsonl"), os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o600) + if err != nil { + return + } + f.Write(append(b, '\n')) + f.Close() + if time.Since(history.pruned) > time.Hour { + history.pruned = time.Now() + pruneHistory(dir, time.Now()) + } +} + +// pruneHistory gzips the days that are over, and drops the days older than +// historyDays, then the oldest until what is left fits historyBytes. +func pruneHistory(dir string, now time.Time) { + today := now.Local().Format(dayForm) + for _, d := range historyFiles(dir) { + if d.day < today && !strings.HasSuffix(d.path, ".gz") { + gzipFile(d.path) + } + } + oldest := now.Local().AddDate(0, 0, -historyDays+1).Format(dayForm) + days := historyFiles(dir) + var total int64 + for _, d := range days { + total += d.size + } + for _, d := range days { // oldest first + if d.day >= oldest && total <= historyBytes || d.day == today { + continue + } + if os.Remove(d.path) == nil { + total -= d.size + } + } +} + +func gzipFile(path string) { + b, err := os.ReadFile(path) + if err != nil { + return + } + var buf bytes.Buffer + z := gzip.NewWriter(&buf) + z.Write(b) + if z.Close() != nil || os.WriteFile(path+".gz", buf.Bytes(), 0o600) != nil { + return + } + os.Remove(path) +} + +type dayFile struct { + day, path string + size int64 +} + +// historyFiles are the kept days, oldest first. +func historyFiles(dir string) []dayFile { + es, _ := os.ReadDir(dir) + var out []dayFile + for _, e := range es { + name := e.Name() + day, _, _ := strings.Cut(name, ".") + if _, err := time.Parse(dayForm, day); err != nil || !(strings.HasSuffix(name, ".jsonl") || strings.HasSuffix(name, ".jsonl.gz")) { + continue + } + info, err := e.Info() + if err != nil { + continue + } + out = append(out, dayFile{day, filepath.Join(dir, name), info.Size()}) + } + sort.Slice(out, func(i, j int) bool { return out[i].day < out[j].day }) + return out +} + +// HistoryDay is a kept day, and how many requests it had. +type HistoryDay struct { + Day string `json:"day"` + Requests int `json:"requests"` +} + +// History is the days kept, newest first, and the routes of day, oldest +// first — its latest historyMax, and whether there were more. +func History(day string) (days []HistoryDay, routes []Route, cut bool) { + history.mu.Lock() + defer history.mu.Unlock() + routes = []Route{} + if history.counts == nil { + history.counts = map[string]counted{} + } + for _, d := range historyFiles(HistoryDir()) { + // a day is read only when it is the one asked for, or it has + // changed since it was counted: thirty days gzipped are not read + // again at every look + c, ok := history.counts[d.path] + if d.day == day || !ok || c.size != d.size { + n := 0 + readDay(d.path, func(line []byte) { + n++ + if d.day != day { + return + } + var r Route + if json.Unmarshal(line, &r) == nil { + routes = append(routes, r) + } + }) + c = counted{d.size, n} + history.counts[d.path] = c + } + if n := c.n; n > 0 { + days = append([]HistoryDay{{d.day, n}}, days...) + } + } + sort.SliceStable(routes, func(i, j int) bool { return routes[i].Time.Before(routes[j].Time) }) + if len(routes) > historyMax { + routes, cut = routes[len(routes)-historyMax:], true + } + return days, routes, cut +} + +// readDay calls f with each line of a day's file. +func readDay(path string, f func([]byte)) { + file, err := os.Open(path) + if err != nil { + return + } + defer file.Close() + var r io.Reader = file + if strings.HasSuffix(path, ".gz") { + z, err := gzip.NewReader(file) + if err != nil { + return + } + defer z.Close() + r = z + } + sc := bufio.NewScanner(r) + sc.Buffer(make([]byte, 64<<10), 16<<20) + for sc.Scan() { + if line := bytes.TrimSpace(sc.Bytes()); len(line) > 0 { + f(line) + } + } +} diff --git a/internal/gateway/history_test.go b/internal/gateway/history_test.go new file mode 100644 index 00000000..56e45fe0 --- /dev/null +++ b/internal/gateway/history_test.go @@ -0,0 +1,52 @@ +package gateway + +import ( + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +// A done route is kept on disk by its day, a day that is over gzipped and +// still read, and a day too old or past the size kept dropped. +func TestHistoryKeepsDaysAndDropsOld(t *testing.T) { + t.Setenv("XDG_CONFIG_HOME", t.TempDir()) + now := time.Now() + yday := now.AddDate(0, 0, -1) + saveRoute(Route{ID: 1, Time: yday, Model: "a", Done: true}) + saveRoute(Route{ID: 2, Time: now, Model: "b", Done: true}) + saveRoute(Route{ID: 3, Time: now.Add(time.Second), Model: "c", Done: true}) + dir := HistoryDir() + old := filepath.Join(dir, now.AddDate(0, 0, -historyDays-2).Format(dayForm)+".jsonl") + os.WriteFile(old, []byte(`{"id":9,"model":"z"}`+"\n"), 0o600) + + pruneHistory(dir, now) + if _, err := os.Stat(old); !os.IsNotExist(err) { + t.Fatal("a day older than the days kept is still there") + } + if _, err := os.Stat(filepath.Join(dir, yday.Format(dayForm)+".jsonl.gz")); err != nil { + t.Fatal("yesterday isn't gzipped:", err) + } + days, rs, _ := History(now.Format(dayForm)) + if len(days) != 2 || days[0].Requests != 2 || days[1].Requests != 1 || len(rs) != 2 || rs[0].Model != "b" || rs[1].Model != "c" { + t.Fatalf("days %+v routes %+v", days, rs) + } + if _, rs, _ := History(yday.Format(dayForm)); len(rs) != 1 || rs[0].Model != "a" { + t.Fatalf("yesterday, gzipped: %+v", rs) + } + // over the bytes kept: the oldest go, today stays + big := strings.Repeat("x", historyBytes/3) + for i := 3; i <= 5; i++ { + os.WriteFile(filepath.Join(dir, now.AddDate(0, 0, -i).Format(dayForm)+".jsonl.gz"), []byte(big), 0o600) + } + pruneHistory(dir, now) + var left []string + for _, d := range historyFiles(dir) { + left = append(left, d.day) + } + want := []string{now.AddDate(0, 0, -4).Format(dayForm), now.AddDate(0, 0, -3).Format(dayForm), yday.Format(dayForm), now.Format(dayForm)} + if strings.Join(left, ",") != strings.Join(want, ",") { + t.Fatalf("kept %v, want %v", left, want) + } +} diff --git a/internal/gateway/trace.go b/internal/gateway/trace.go index 4ad8d06a..bfe42124 100644 --- a/internal/gateway/trace.go +++ b/internal/gateway/trace.go @@ -212,6 +212,11 @@ func (t *trace) changed() { func (t *trace) begin(r Route) *Route { t.mu.Lock() defer t.mu.Unlock() + if t.ids == 0 { + // ids go on from one run to the next, the history keeping them + // all: counted on from the time the gateway began + t.ids = time.Now().UnixMilli() + } t.ids++ r.ID = t.ids if r.Tries == nil { @@ -246,6 +251,11 @@ func (t *trace) update(r *Route, f func(r *Route)) { if r.Status >= 400 { t.totals.Errors++ } + c := *r + c.Order = append([]Weighed(nil), r.Order...) + c.Left = append([]Weighed(nil), r.Left...) + c.Tries = append([]Try{}, r.Tries...) + go saveRoute(c) } t.changed() r.Seq = t.seq diff --git a/internal/gui/assets/i18n.js b/internal/gui/assets/i18n.js index 208709a9..0e8157e1 100644 --- a/internal/gui/assets/i18n.js +++ b/internal/gui/assets/i18n.js @@ -931,6 +931,11 @@ const I18N = { "request {i} of {n}": "第 {i}/{n} 个请求", "sent {time}": "发出于 {time}", "starting…": "即将开始…", + "Replay from here": "从这里重放", + "Live": "实时", + "yesterday": "昨天", + "{n} on {day}": "{day}的 {n} 个请求", + "the last {n} of {day}": "{day}最后的 {n} 个请求", "How the last request was routed": "最近一个请求是怎么路由的", "How the request at {time} was routed": "{time} 的请求是怎么路由的", "Waiting for a request": "等待请求", diff --git a/internal/gui/assets/routing.css b/internal/gui/assets/routing.css index b8e0011a..355c30d2 100644 --- a/internal/gui/assets/routing.css +++ b/internal/gui/assets/routing.css @@ -304,3 +304,11 @@ .rp-mark.on { opacity: 1; transform: scaleY(1.5); } .rp-head { position: absolute; top: 1px; width: 2px; height: 14px; margin-left: -1px; border-radius: 1px; background: var(--fg); pointer-events: none; } @media (max-width: 520px) { .rp-req .rp-at { display: none; } } + +/* the days the routing history keeps: live, then each day, newest first */ +.rt-days { display: flex; gap: 4px; overflow-x: auto; padding: 0 2px 6px; scrollbar-width: none; } +.rt-days::-webkit-scrollbar { display: none; } +.rt-day { flex: none; display: inline-flex; align-items: baseline; gap: 5px; font-size: 12px; color: var(--muted); padding: 3px 9px; border: 1px solid var(--line); border-radius: 999px; background: none; white-space: nowrap; } +.rt-day small { font-size: 10.5px; color: var(--faint); font-variant-numeric: tabular-nums; } +.rt-day:hover { color: var(--fg); } +.rt-day.on { color: var(--fg); border-color: var(--fg); } diff --git a/internal/gui/assets/routing.js b/internal/gui/assets/routing.js index 698a44b9..eaeca479 100644 --- a/internal/gui/assets/routing.js +++ b/internal/gui/assets/routing.js @@ -81,11 +81,13 @@ const more = $("#rtMore"); const reqHead = el("div", "row-head"), reqNote = el("span", "note"); const reqs = el("div", "list rt-reqs"); + // the days the history keeps on disk, to look back at one: see listed + const dayBar = el("div", "rt-days"); const actHead = el("div", "row-head"), actNote = el("span", "note"); const acts = el("div", "list rt-acts"); const hist = el("div", "rt-cols"); const colA = el("div", "rt-col"), colB = el("div", "rt-col"); - colA.append(reqHead, reqs); + colA.append(reqHead, dayBar, reqs); colB.append(actHead, acts); hist.append(colA, colB); more.append(hist); @@ -442,7 +444,7 @@ // ---------- state ---------- const routes = new Map(); // id → the latest of each route - let seq = 0, mine = true, loaded = false; + let seq = 0, mine = true, loaded = false, daysAt = 0; let cur = null; // the route the header and the log tell of: the newest played let pinned = null; // a past route picked from the strip let rows = new Map(); // id → { li, wire, st, bi, tg, w, rid, up } @@ -451,6 +453,7 @@ let sets = []; // the account sets on the stage, in the order they came const playing = new Map(); // the routes being played → the gen playing each let rp = null; // a replay playing: see replay + let day = "", days = [], past = [], pastCut = false; // a kept day looked at ("" for live), the days kept, its routes const src = () => rp ? rp.routes : routes; // the routes the stage plays from const LINGER = 12e3; // how long an agent's last request stays on the stage let gen = 0, trips = [], waiters = []; @@ -851,10 +854,17 @@ const again = el("button", "text", t("Replay")); again.onclick = () => replay([r], pinned); logHead.append(again); + // and on from it: the requests listed after it, as they came + const on = listed().filter((x) => x.id >= r.id); + if (pinned && on.length > 1) { + const from = el("button", "text", t("Replay from here")); + from.onclick = () => replay(on, pinned); + logHead.append(from); + } } if (pinned && !rp) { const live = el("button", "text", t("Back to live")); - live.onclick = () => { pinned = null; cur = newest(); sync(true); renderAll(); }; + live.onclick = () => { if (day) lookAt(""); else { pinned = null; cur = newest(); sync(true); renderAll(); } }; logHead.append(live); } const items = []; @@ -896,16 +906,59 @@ } // the requests the gateway keeps, newest first: pick one to see how it was routed + // listed: the requests the list shows, newest first — the gateway's last + // few, or a day the history keeps + const listed = () => (day ? past : [...routes.values()]).slice().sort((a, b) => b.id - a.id); + const dayName = (d) => { + const x = new Date(d + "T12:00:00"), n = new Date(), y = new Date(n.getTime() - 864e5); + return x.toDateString() === n.toDateString() ? t("today") : x.toDateString() === y.toDateString() ? t("yesterday") + : x.toLocaleDateString([], { month: "short", day: "numeric", weekday: "short" }); + }; + async function loadDays(d) { + try { + const res = await (await fetch("/api/gateway/history?day=" + encodeURIComponent(d || ""))).json(); + days = res.days || []; + if (d && d === day) { past = res.routes || []; pastCut = !!res.cut; } + } catch {} + renderDays(); + } + async function lookAt(d) { + if (rp) endReplay(true); + day = d; + past = []; + if (d) await loadDays(d); + pinned = null; + const r = d ? listed()[0] : newest(); + if (r && d) pick(r); + else if (r) { stopPlays(); cur = r; sync(true); say(affWhy(r, true) || ruleWhy(r, true) || firstWhy(r)); } + renderAll(); + } + function renderDays() { + const b = (d, label, n) => { + const x = el("button", "rt-day" + (d === day ? " on" : "")); + x.append(el("span", "", label), ...(n != null ? [el("small", "", String(n))] : [])); + x.setAttribute("aria-pressed", String(d === day)); + x.onclick = () => { if (d !== day) lookAt(d); }; + return x; + }; + const key = day + "|" + days.map((d) => d.day + ":" + d.requests).join(","); + if (dayBar.dataset.key === key) return; + dayBar.dataset.key = key; + dayBar.replaceChildren(b("", t("Live")), ...days.map((d) => b(d.day, dayName(d.day), d.requests))); + dayBar.hidden = !days.length && !day; + } function renderHist() { - const rs = [...routes.values()].sort((a, b) => b.id - a.id); - hist.hidden = !rs.length; + const rs = listed(); + hist.hidden = !rs.length && !day && !days.length; reqHead.replaceChildren(el("span", "label", t("Requests")), el("span", "grow"), reqNote); if (rs.filter((r) => r.done).length > 1 && !rp) { const all = el("button", "text", t("Replay them all")); all.onclick = () => replay(rs, pinned); reqHead.append(all); } - reqNote.textContent = t("the last {n} the gateway keeps", { n: rs.length }); + reqNote.textContent = day ? t(pastCut ? "the last {n} of {day}" : "{n} on {day}", { n: rs.length, day: dayName(day) }) + : t("the last {n} the gateway keeps", { n: rs.length }); + renderDays(); reqs.replaceChildren(...rs.map((r) => { const [said, how, tr] = outcome(r); const b = el("button", "rt-req " + how); @@ -1319,7 +1372,7 @@ rp = null; rbar.hidden = true; if (stop) stopPlays(); - const b = p.back && routes.get(p.back.id); + const b = p.back && (routes.get(p.back.id) || (day && past.find((x) => x.id === p.back.id))); pinned = b || null; cur = b || newest(); if (cur) { sync(true); renderAll(); } @@ -1383,7 +1436,7 @@ requestAnimationFrame(frame); } // countdowns tick once a second - setInterval(() => { if (shown()) { if (!pinned && loaded && cur) sync(); render(); renderActs([...routes.values()].sort((a, b) => b.id - a.id)); } }, 1000); + setInterval(() => { if (shown()) { if (!pinned && loaded && cur) sync(); render(); renderActs(listed()); } }, 1000); function offline(msg) { off.textContent = msg; @@ -1446,6 +1499,8 @@ } for (const id of [...routes.keys()].sort((a, b) => a - b).slice(0, -60)) routes.delete(id); loaded = true; + // a request done is a day's count grown: heard of now and then + if (first || (d.routes.some((r) => r.done) && performance.now() - daysAt > 15e3)) { daysAt = performance.now(); loadDays(); } if (first) { offline(""); // the agents' names come with the app's state, which may not be here yet diff --git a/internal/gui/trace.go b/internal/gui/trace.go index 3a7528c2..d4337946 100644 --- a/internal/gui/trace.go +++ b/internal/gui/trace.go @@ -34,6 +34,12 @@ func traceRoutes(mux *http.ServeMux) { out.Now = time.Now() writeJSON(rw, out) }) + // the routes of a day gone by, from the history the gateway keeps on + // disk — read whichever magpie serves the gateway + mux.HandleFunc("GET /api/gateway/history", func(rw http.ResponseWriter, r *http.Request) { + days, routes, cut := gateway.History(r.URL.Query().Get("day")) + writeJSON(rw, map[string]any{"days": days, "routes": routes, "cut": cut}) + }) // an account's rest lifted by hand: verified with its vendor, say mux.HandleFunc("POST /api/gateway/unrest", func(rw http.ResponseWriter, r *http.Request) { var in struct {