package main
import (
"context"
"errors"
"log"
"time"
"craftdesign.group/quarry/internal/errlog"
"craftdesign.group/quarry/internal/fault"
"craftdesign.group/quarry/internal/ingest"
"craftdesign.group/quarry/internal/work"
)
// ingestShedDeadline is how long a store's poll rounds may keep being
// shed at the ingest lane before the journal hears of it. A shed is
// back-pressure: while boot rebuilds hold the lane, every other store is
// refused for a while and then gets in, and journalling each refusal
// made 613 "a store could not be ingested" entries in minutes out of an
// outage that lost nothing. A store still refused after this long is a
// fact an operator needs.
const ingestShedDeadline = 2 * time.Minute
// maxShedBackoff caps the wait between a store's shed rounds. The lane's
// own shed counter (Refused on the operations page) counts each one.
const maxShedBackoff = 30 * time.Second
// shedBackoff is one store's run of rounds shed at the ingest lane.
// Owned by that store's poll loop.
type shedBackoff struct {
since, retryAt time.Time
wait time.Duration
journalled bool
}
// ingestRound runs one poll round admitted through the ingest lane; a
// round that turns out to be a rebuild leaves the lane (ingest.Admit).
// ran is false when the round was not run: still inside its backoff, or
// shed now. Neither is an error; a store shed past ingestShedDeadline is
// journalled once, here.
func (rg *registrar) ingestRound(ctx context.Context, now time.Time, name string, shed *shedBackoff, fn func(ingest.Admit) error) (ran bool, err error) {
if now.Before(shed.retryAt) {
return false, nil
}
// Shed means the lane refused THIS round. An ErrBusy from inside an
// admitted round is another lane's refusal, and that round ran.
admitted := false
err = fn(func(body func(leave func()) error) error {
return rg.ingestLane.DoNamedLeavable(ctx, "ingest", name, func(leave func()) error {
admitted = true
return body(leave)
})
})
if admitted || !errors.Is(err, work.ErrBusy) {
*shed = shedBackoff{}
return true, err
}
if shed.since.IsZero() {
shed.since = now
}
shed.wait = min(max(2*shed.wait, rg.interval), maxShedBackoff)
shed.retryAt = now.Add(shed.wait)
if !shed.journalled && now.Sub(shed.since) >= ingestShedDeadline {
shed.journalled = true
log.Printf("ingest %s: refused at the ingest lane for %s: %v", name, now.Sub(shed.since).Round(time.Second), err)
errlog.Default.ReportIn("ingest", "ingest "+name, name, "",
fault.Wrap(err, "a store's ingest rounds have been refused for longer than the deadline").
Cost("this store's new check-ins are not searchable or listed until a round gets in; the store itself is intact").
With("repo", name, "refused_for", now.Sub(shed.since).Round(time.Second).String()))
}
return false, nil
}