diff options
| -rw-r--r-- | README.md | 3 | ||||
| -rw-r--r-- | cmds/rss.go | 2 | ||||
| -rw-r--r-- | cmds/rss_test.go | 2 | ||||
| -rw-r--r-- | cmds/serve.go | 80 | ||||
| -rw-r--r-- | cmds/serve_test.go | 57 | ||||
| -rw-r--r-- | justfile | 6 |
6 files changed, 135 insertions, 15 deletions
@@ -49,6 +49,9 @@ runs.jsonl server run log; doubles as the RSS render source ## Notes +- Serve surfaces: `GET /recommendations.xml`, `GET /api/status`, and + `POST /api/poll` (out-of-schedule poll; serialized with scheduled polls, + short cooldown, `curl -X POST https://server/api/poll`). - Never writes to Miniflux. Never needs credentials in the repo. - Training stays on the local PC; the server only scores (low RAM). - Deploy `serve` only after a `train` run on enough organic stars shows an diff --git a/cmds/rss.go b/cmds/rss.go index 30be578..f5f28a9 100644 --- a/cmds/rss.go +++ b/cmds/rss.go @@ -56,7 +56,7 @@ func RenderRSS(runs []core.RunRow) ([]byte, error) { doc := rssDocument{ Version: "2.0", Channel: rssChannel{ - Title: "fluxrec recommendations", + Title: "Fluxrecs", Description: "Top-scored Miniflux entries, refreshed on a schedule.", }, } diff --git a/cmds/rss_test.go b/cmds/rss_test.go index 79d6d37..5ff3405 100644 --- a/cmds/rss_test.go +++ b/cmds/rss_test.go @@ -44,7 +44,7 @@ func TestRenderRSSGolden(t *testing.T) { want := `<?xml version="1.0" encoding="UTF-8"?> <rss version="2.0"> <channel> - <title>fluxrec recommendations</title> + <title>Fluxrecs</title> <description>Top-scored Miniflux entries, refreshed on a schedule.</description> <item> <title>Gamma — Feed A</title> diff --git a/cmds/serve.go b/cmds/serve.go index 0df5003..bba643a 100644 --- a/cmds/serve.go +++ b/cmds/serve.go @@ -24,6 +24,7 @@ import ( "sort" "strconv" "strings" + "sync" "sync/atomic" "syscall" "time" @@ -56,9 +57,11 @@ func (c *ServeCommand) Init(args []string) error { Poll Miniflux for new entries, score each once, keep a top-N + exploration batch per refresh in runs.jsonl, and serve the recent batches as RSS at -/recommendations.xml plus JSON status at /api/status. Read-only toward -Miniflux. Refresh times are wall-clock in the process timezone ($TZ, -default UTC). Exits immediately if the model file is missing or corrupt. +/recommendations.xml plus JSON status at /api/status. POST /api/poll +triggers an out-of-schedule poll (serialized with scheduled polls, brief +cooldown). Read-only toward Miniflux. Refresh times are wall-clock in the +process timezone ($TZ, default UTC). Exits immediately if the model file +is missing or corrupt. `) fs.PrintDefaults() } @@ -104,13 +107,23 @@ func (c *ServeCommand) Run(stdin io.Reader, stdout io.Writer) error { mins, _ := parseRefreshAt(c.RefreshAt) // validated in Init rng := rand.New(rand.NewSource(time.Now().UnixNano())) - poll := func() { + // Serialized: a manual POST /api/poll and a scheduled refresh must never + // overlap (both read and advance the same cursor, so a race would + // double-score a batch). TryLock makes the loser fail fast — the + // scheduler simply retries next cycle. + var pollMu sync.Mutex + poll := func() error { + if !pollMu.TryLock() { + return errors.New("a poll is already running") + } + defer pollMu.Unlock() if err := c.pollOnce(client, vec, model, rng, stdout); err != nil { c.pollsFailed.Add(1) fmt.Fprintf(os.Stderr, "serve: poll failed (cursor not advanced, will retry next cycle): %v\n", err) - } else { - c.pollsOK.Add(1) + return err } + c.pollsOK.Add(1) + return nil } // Boot catch-up: if the last logged run is older than half the smallest @@ -121,7 +134,7 @@ func (c *ServeCommand) Run(stdin io.Reader, stdout io.Writer) error { return err } if due { - poll() + _ = poll() } ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) @@ -129,7 +142,7 @@ func (c *ServeCommand) Run(stdin io.Reader, stdout io.Writer) error { srv := &http.Server{ Addr: c.Addr, - Handler: c.routes(mins, model.CreatedAt), + Handler: c.routes(mins, model.CreatedAt, poll), ReadTimeout: core.DefaultReadTimeout, WriteTimeout: core.DefaultWriteTimeout, IdleTimeout: core.DefaultIdleTimeout, @@ -149,7 +162,7 @@ func (c *ServeCommand) Run(stdin io.Reader, stdout io.Writer) error { timer.Stop() return case <-timer.C: - poll() + _ = poll() } } }() @@ -425,9 +438,54 @@ func minGapMinutes(mins []int) int { return gap } -// routes wires the two surfaces: the RSS feed and the JSON status. -func (c *ServeCommand) routes(mins []int, modelCreatedAt time.Time) http.Handler { +// Cooldown between polls to gate the public manual-trigger endpoint: a +// run less than this old means the last cycle is still fresh enough. +const minManualPollGap = time.Minute + +// routes wires the three surfaces: the RSS feed, the JSON status, and the +// manual poll trigger. POST /api/poll runs the same serialized poll the +// scheduler uses (the passed-in pollFn refuses to overlap; a poll that +// just ran also gets a 429 so the endpoint can't drive load). It's +// unauthenticated like the other surfaces: the worst it can cause is a +// read-only fetch+score cycle against Miniflux. +func (c *ServeCommand) routes(mins []int, modelCreatedAt time.Time, pollFn func() error) http.Handler { mux := http.NewServeMux() + mux.HandleFunc("/api/poll", func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + w.Header().Set("Allow", http.MethodPost) + http.Error(w, "use POST", http.StatusMethodNotAllowed) + return + } + before, err := core.ReadRunsTail(c.Runs, 1) + if err != nil { + http.Error(w, "reading runs: "+err.Error(), http.StatusInternalServerError) + return + } + if len(before) > 0 && time.Since(before[0].RanAt) < minManualPollGap { + http.Error(w, "a poll ran less than a minute ago; try later", http.StatusTooManyRequests) + return + } + if err := pollFn(); err != nil { + http.Error(w, "poll failed: "+err.Error(), http.StatusBadGateway) + return + } + after, err := core.ReadRunsTail(c.Runs, 1) + if err != nil { + http.Error(w, "reading runs: "+err.Error(), http.StatusInternalServerError) + return + } + newRun := len(after) > 0 && (len(before) == 0 || !after[len(after)-1].RanAt.Equal(before[len(before)-1].RanAt)) + resp := map[string]any{"polled": true, "new_run": newRun} + if newRun { + row := after[len(after)-1] + resp["ran_at"] = row.RanAt + resp["entries_seen"] = row.EntriesSeen + resp["kept"] = len(row.Kept) + resp["score_median"] = row.ScoreMedian + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(resp) + }) mux.HandleFunc("/recommendations.xml", func(w http.ResponseWriter, r *http.Request) { runs, err := core.ReadRunsTail(c.Runs, c.RSSBatches) if err != nil { diff --git a/cmds/serve_test.go b/cmds/serve_test.go index e8aec19..509de6c 100644 --- a/cmds/serve_test.go +++ b/cmds/serve_test.go @@ -384,7 +384,9 @@ func TestServeRoutes(t *testing.T) { } cmd := &ServeCommand{Runs: runsPath, RSSBatches: 21} - handler := cmd.routes([]int{360, 720, 1080}, time.Now().Add(-48*time.Hour)) + pollCalls := 0 + pollStub := func() error { pollCalls++; return nil } + handler := cmd.routes([]int{360, 720, 1080}, time.Now().Add(-48*time.Hour), pollStub) rec := httptest.NewRecorder() handler.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/recommendations.xml", nil)) @@ -422,3 +424,56 @@ func TestServeRoutes(t *testing.T) { t.Errorf("model_age_hours = %v, want positive", status["model_age_hours"]) } } + +func TestManualPollRoute(t *testing.T) { + dir := t.TempDir() + runsPath := filepath.Join(dir, "runs.jsonl") + // Last run hours ago: well outside the manual-poll cooldown. + if err := core.AppendRun(runsPath, core.RunRow{ + RanAt: time.Now().Add(-time.Hour), EntriesSeen: 3, + }); err != nil { + t.Fatalf("AppendRun: %v", err) + } + + pollCalls := 0 + pollStub := func() error { pollCalls++; return nil } + cmd := &ServeCommand{Runs: runsPath} + handler := cmd.routes([]int{360, 720, 1080}, time.Now(), pollStub) + + // GET is rejected; only POST triggers. + rec := httptest.NewRecorder() + handler.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/api/poll", nil)) + if rec.Code != http.StatusMethodNotAllowed { + t.Errorf("GET /api/poll = %d, want 405", rec.Code) + } + + // POST runs the poll; the stub wrote no row, so no new run is reported. + rec = httptest.NewRecorder() + handler.ServeHTTP(rec, httptest.NewRequest(http.MethodPost, "/api/poll", nil)) + if rec.Code != http.StatusOK { + t.Fatalf("POST /api/poll = %d, want 200", rec.Code) + } + if pollCalls != 1 { + t.Errorf("poll called %d times, want 1", pollCalls) + } + var resp map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil { + t.Fatalf("response not JSON: %v", err) + } + if resp["polled"] != true || resp["new_run"] != false { + t.Errorf("response = %v, want polled=true new_run=false", resp) + } + + // A fresh run row puts the next POST inside the cooldown window. + if err := core.AppendRun(runsPath, core.RunRow{RanAt: time.Now(), EntriesSeen: 5}); err != nil { + t.Fatalf("AppendRun: %v", err) + } + rec = httptest.NewRecorder() + handler.ServeHTTP(rec, httptest.NewRequest(http.MethodPost, "/api/poll", nil)) + if rec.Code != http.StatusTooManyRequests { + t.Errorf("POST inside cooldown = %d, want 429", rec.Code) + } + if pollCalls != 1 { + t.Errorf("poll called %d times, want still 1 (cooldown)", pollCalls) + } +} @@ -64,4 +64,8 @@ restart SERVER: # deploy + restart in one verb ship SERVER: (deploy SERVER) (restart SERVER) - @echo "model shipped; serve restarted — watch /api/status next run"
\ No newline at end of file + @echo "model shipped; serve restarted — watch /api/status next run" + +# Trigger an out-of-schedule poll (serialized + cooldown-gated server-side) +poll-now URL: + curl -sS -X POST {{URL}}/api/poll
\ No newline at end of file |