orchard

  1package main
  2
  3import (
  4	"context"
  5	"encoding/json"
  6	"errors"
  7	"fmt"
  8	"log/slog"
  9	"os"
 10	"path/filepath"
 11	"sync"
 12	"time"
 13)
 14
 15// Every upstream this site reads is free, keyless and someone else's. A ban on
 16// one of them is not something a retry fixes and it lands on Isaac rather than
 17// on a reviewer, so each endpoint is fenced: a hard budget per hour, a pause
 18// between consecutive calls, and a breaker that opens on the responses that
 19// mean stop asking.
 20//
 21// The state is written to disk because a breaker that resets on restart is not
 22// a breaker. A restart loop against an endpoint that just returned 429 is
 23// exactly the case this is here to prevent, and an in-memory counter would
 24// forget about it every few seconds. It is JSON rather than SQLite because
 25// four counters per endpoint do not need a database, and skipping one keeps
 26// this the only site in the repo with no cgo-free driver, no volume of
 27// consequence and no schema to migrate.
 28
 29var errGuardOpen = errors.New("circuit open")
 30
 31// Budgets are per rolling hour and are ceilings to fail against, not targets.
 32// The pollers ask for far less: markets at 30s is 120/hr, and everything else
 33// is minutes apart.
 34type budget struct {
 35	perHour int
 36	pace    time.Duration
 37}
 38
 39var budgets = map[string]budget{
 40	"yahoo":     {perHour: 900, pace: 2 * time.Second},
 41	"algolia":   {perHour: 120, pace: 2 * time.Second},
 42	"lobsters":  {perHour: 120, pace: 2 * time.Second},
 43	"openmeteo": {perHour: 120, pace: 2 * time.Second},
 44	"npr":       {perHour: 120, pace: 2 * time.Second},
 45	"bbc":       {perHour: 60, pace: 2 * time.Second},
 46	"nasdaq":    {perHour: 120, pace: 2 * time.Second},
 47	"nws":       {perHour: 120, pace: 2 * time.Second},
 48	"pollen":    {perHour: 60, pace: 2 * time.Second},
 49	"steam":     {perHour: 240, pace: time.Second},
 50	"justwatch": {perHour: 60, pace: 2 * time.Second},
 51	// No pacing on these two. They go to Isaac's own machine rather than to a
 52	// stranger's endpoint, and the strip fires every probe at once, so a
 53	// pause between them would only refuse all but the first.
 54	"uptime":  {perHour: 900},
 55	"logging": {perHour: 900},
 56}
 57
 58const (
 59	// Trip after this many consecutive failures. A single timeout on a free
 60	// endpoint is normal and should not stop the panel.
 61	failsToTrip = 4
 62
 63	minBackoff = 30 * time.Second
 64	maxBackoff = 30 * time.Minute
 65)
 66
 67type guardEntry struct {
 68	WindowStart time.Time     `json:"window_start"`
 69	Count       int           `json:"count"`
 70	Fails       int           `json:"fails"`
 71	LastCall    time.Time     `json:"last_call"`
 72	OpenUntil   time.Time     `json:"open_until"`
 73	Backoff     time.Duration `json:"backoff"`
 74}
 75
 76// Guard is safe for concurrent use. Every poller shares one.
 77type Guard struct {
 78	path string
 79
 80	mu      sync.Mutex
 81	entries map[string]*guardEntry
 82	dirty   bool
 83}
 84
 85func NewGuard(dataDir string) *Guard {
 86	g := &Guard{
 87		path:    filepath.Join(dataDir, "guard.json"),
 88		entries: map[string]*guardEntry{},
 89	}
 90
 91	b, err := os.ReadFile(g.path)
 92	if err != nil {
 93		// A missing file is the first boot. Anything else is worth saying out
 94		// loud, since it means the breaker starts closed against an endpoint
 95		// that may have been the reason for the restart.
 96		if !errors.Is(err, os.ErrNotExist) {
 97			slog.Warn("guard state unreadable, starting closed",
 98				slog.String("component", "guard"), slog.Any("err", err))
 99		}
100		return g
101	}
102	if err := json.Unmarshal(b, &g.entries); err != nil {
103		slog.Warn("guard state unparseable, starting closed",
104			slog.String("component", "guard"), slog.Any("err", err))
105		g.entries = map[string]*guardEntry{}
106	}
107	return g
108}
109
110func (g *Guard) entry(name string) *guardEntry {
111	e, ok := g.entries[name]
112	if !ok {
113		e = &guardEntry{}
114		g.entries[name] = e
115	}
116	return e
117}
118
119// Allow reports whether a call to name may go out now, and reserves a slot in
120// the budget when it may. A caller that is refused must not make the request.
121//
122// Pacing is reported as a refusal here, which is right for a probe that has
123// something better to do than queue. Anything on a timer should call Reserve
124// instead: three pollers sharing the Yahoo endpoint all fired at boot, two were
125// refused for being 1.6 seconds early, and because their retry interval is an
126// hour and six hours the panels they fill stayed empty for the rest of the day.
127func (g *Guard) Allow(name string) error {
128	wait, err := g.tryReserve(name)
129	if err != nil {
130		return err
131	}
132	if wait > 0 {
133		return fmt.Errorf("%s: paced, %s early", name, wait.Round(time.Millisecond))
134	}
135	return nil
136}
137
138// Reserve waits out the pace and then reserves a slot. An open breaker and a
139// spent budget still refuse outright, because neither is fixed by waiting a
140// moment and a caller that queued on them would pile up behind a dead endpoint.
141func (g *Guard) Reserve(ctx context.Context, name string) error {
142	for {
143		wait, err := g.tryReserve(name)
144		if err != nil {
145			return err
146		}
147		if wait == 0 {
148			return nil
149		}
150
151		select {
152		case <-ctx.Done():
153			return ctx.Err()
154		case <-time.After(wait):
155		}
156	}
157}
158
159// tryReserve returns how long the caller has to wait for the pace, or an error
160// for the two conditions that waiting does not help. A zero wait with no error
161// means the slot is taken.
162func (g *Guard) tryReserve(name string) (time.Duration, error) {
163	now := time.Now()
164
165	g.mu.Lock()
166	defer g.mu.Unlock()
167
168	e := g.entry(name)
169
170	if now.Before(e.OpenUntil) {
171		return 0, fmt.Errorf("%s: %w for another %s", name, errGuardOpen, e.OpenUntil.Sub(now).Round(time.Second))
172	}
173
174	b, ok := budgets[name]
175	if !ok {
176		return 0, fmt.Errorf("%s: no budget defined", name)
177	}
178
179	if now.Sub(e.WindowStart) >= time.Hour {
180		e.WindowStart = now
181		e.Count = 0
182	}
183	if e.Count >= b.perHour {
184		return 0, fmt.Errorf("%s: hourly budget of %d spent", name, b.perHour)
185	}
186	if since := now.Sub(e.LastCall); since < b.pace {
187		return b.pace - since, nil
188	}
189
190	e.Count++
191	e.LastCall = now
192	g.dirty = true
193	return 0, nil
194}
195
196// Succeed closes the breaker and clears the backoff.
197func (g *Guard) Succeed(name string) {
198	g.mu.Lock()
199	defer g.mu.Unlock()
200
201	e := g.entry(name)
202	if e.Fails != 0 || e.Backoff != 0 {
203		e.Fails = 0
204		e.Backoff = 0
205		g.dirty = true
206	}
207}
208
209// Fail records an unsuccessful call. status is the HTTP status where there was
210// one and 0 for a transport error. 429 and 503 open the breaker at once,
211// because they are the endpoint saying so rather than the network being bad.
212func (g *Guard) Fail(name string, status int, retryAfter time.Duration) {
213	g.mu.Lock()
214	defer g.mu.Unlock()
215
216	e := g.entry(name)
217	e.Fails++
218	g.dirty = true
219
220	immediate := status == 429 || status == 503
221	if !immediate && e.Fails < failsToTrip {
222		return
223	}
224
225	if e.Backoff == 0 {
226		e.Backoff = minBackoff
227	} else {
228		e.Backoff *= 2
229	}
230	if e.Backoff > maxBackoff {
231		e.Backoff = maxBackoff
232	}
233	// An explicit Retry-After wins over the doubling, both ways: the endpoint
234	// knows when it wants us back and guessing shorter is how a soft limit
235	// becomes a hard one.
236	wait := e.Backoff
237	if retryAfter > wait {
238		wait = retryAfter
239	}
240
241	e.OpenUntil = time.Now().Add(wait)
242	slog.Warn("upstream breaker opened",
243		slog.String("component", "guard"),
244		slog.String("endpoint", name),
245		slog.Int("status", status),
246		slog.String("for", wait.String()))
247}
248
249// Flush writes the state out when it has changed. Called on a timer rather
250// than on every call so a 30 second poll does not mean a write per tick.
251func (g *Guard) Flush() {
252	g.mu.Lock()
253	if !g.dirty {
254		g.mu.Unlock()
255		return
256	}
257	b, err := json.Marshal(g.entries)
258	g.dirty = false
259	g.mu.Unlock()
260
261	if err != nil {
262		slog.Warn("guard state unmarshalable", slog.String("component", "guard"), slog.Any("err", err))
263		return
264	}
265	// Rename onto the real path so a crash mid-write leaves the old state
266	// rather than a truncated file the next boot would refuse to parse.
267	tmp := g.path + ".tmp"
268	if err := os.WriteFile(tmp, b, 0o644); err != nil {
269		slog.Warn("guard state unwritable", slog.String("component", "guard"), slog.Any("err", err))
270		return
271	}
272	if err := os.Rename(tmp, g.path); err != nil {
273		slog.Warn("guard state not renamed", slog.String("component", "guard"), slog.Any("err", err))
274	}
275}
276
277// Feed is one upstream as the SIGNAL panel shows it.
278type Feed struct {
279	Name  string `json:"name"`
280	State string `json:"state"`
281	Age   string `json:"age"`
282
283	// Calls taken out of this hour's budget, and the budget itself, so the
284	// panel answers whether anything here is being hammered rather than
285	// leaving it to be worked out from the poll intervals in the source.
286	Used   int `json:"used"`
287	Budget int `json:"budget"`
288	Load   int `json:"load"`
289}
290
291// Feeds reports every upstream this site reads, in a fixed order so the panel
292// does not reshuffle between polls the way a map range would.
293var feedOrder = []struct{ key, label string }{
294	{"yahoo", "YAHOO"},
295	{"algolia", "HN"},
296	{"lobsters", "LOBSTERS"},
297	{"npr", "NPR"},
298	{"bbc", "BBC"},
299	{"nasdaq", "NASDAQ"},
300	{"openmeteo", "OPEN-METEO"},
301	{"nws", "NWS"},
302	{"pollen", "POLLEN"},
303	{"steam", "STEAM"},
304	{"justwatch", "JUSTWATCH"},
305	{"logging", "LOGGING"},
306}
307
308func (g *Guard) Feeds(now time.Time) []Feed {
309	order := feedOrder
310
311	g.mu.Lock()
312	defer g.mu.Unlock()
313
314	feeds := make([]Feed, 0, len(order))
315	for _, o := range order {
316		f := Feed{Name: o.label, State: "idle", Age: "--"}
317		if b, ok := budgets[o.key]; ok {
318			f.Budget = b.perHour
319		}
320		if e, ok := g.entries[o.key]; ok {
321			// The window rolls, so a count from a window that has already
322			// expired is not this hour's usage.
323			if now.Sub(e.WindowStart) < time.Hour {
324				f.Used = e.Count
325				if f.Budget > 0 {
326					f.Load = f.Used * 100 / f.Budget
327				}
328			}
329			switch {
330			case now.Before(e.OpenUntil):
331				f.State = "open"
332			case e.Fails > 0:
333				f.State = "degraded"
334			default:
335				f.State = "ok"
336			}
337			if !e.LastCall.IsZero() {
338				f.Age = shortAge(now.Sub(e.LastCall))
339			}
340		}
341		feeds = append(feeds, f)
342	}
343	return feeds
344}
345
346func shortAge(d time.Duration) string {
347	switch {
348	case d < time.Minute:
349		return fmt.Sprintf("%ds", int(d.Seconds()))
350	case d < time.Hour:
351		return fmt.Sprintf("%dm", int(d.Minutes()))
352	default:
353		return fmt.Sprintf("%dh", int(d.Hours()))
354	}
355}
356
357// Status is what the footer shows: which endpoints are currently shut off.
358func (g *Guard) Status() []string {
359	now := time.Now()
360
361	g.mu.Lock()
362	defer g.mu.Unlock()
363
364	var open []string
365	for name, e := range g.entries {
366		if now.Before(e.OpenUntil) {
367			open = append(open, name)
368		}
369	}
370	return open
371}