Added a backfill in 7-day windows, and fetches now keep their data on a failure

Backfilling a long window (an early --since, or a long time offline)
hit the API's rate limit, which could still be exceeded after waiting
once, and then everything fetched so far was lost.

Before each Find a Tender source's worker starts, a start point more
than 7 days ago is now backfilled in 7-day windows bounded on both
sides, oldest first, each saved before the next is fetched. A window
that fails is saved as far as it got and fetched again from its start
after the source's interval. The worker then carries on from the end
of the last window.

A 429 is now waited out up to sources.<source>.rate_limit_retries
times (default 5) rather than once; fts-once --ignore-rate-limit uses
0. FetchAll returns what it fetched before an error along with it, and
the workers save it, so a failed run no longer loses its data.
findatender's FetchAll and FetchOnce take RequestOptions, which gains
Until (updatedTo), and RateLimitRetries replaces AllowRateLimit.

README.md

@@ -118,6 +118,20 @@ Without it, `service` (and `job-search serve`) starts each source from the
 newest release stored for it, so a restart carries on where it left off; with
 nothing stored, Find a Tender's client fetches the last day. `--since`
 overrides that for every source.
+
+When the start is more than 7 days ago (an early `--since`, or a long time
+offline), the Find a Tender sources first backfill it in 7-day windows,
+oldest first, saving each before fetching the next. A window that fails is
+saved as far as it got and fetched again after the source's interval. The
+API returns the newest releases first, so if a backfill is stopped partway
+through a window, restarting without `--since` skips the rest of that
+window: restart with `--since` set to the window's start (logged as
+`backfilling`) to fill it.
+
+A rate limit is waited out, as the API asks, up to
+`sources.<source>.rate_limit_retries` times per request. A failed fetch
+keeps what it fetched before the failure.
+
 `fts-once` can be limited with `--planning`, `--opportunities` or
 `--awards`, and with `--ignore-rate-limit` it fails on a rate limit instead
 of waiting.

internal/cli/fetch/fts_once.go

@@ -83,8 +83,15 @@ func commandFTSOnce(opts *cli.Options) *cobra.Command {
 			defer findATender.Shutdown()
 
 			ignoreRateLimit, _ := cmd.Flags().GetBool("ignore-rate-limit")
+			rateLimitRetries := conf.Sources.FindATender.RateLimitRetries
+			if ignoreRateLimit {
+				rateLimitRetries = 0
+			}
 
-			results, err := findATender.FetchAll(since, !ignoreRateLimit)
+			results, err := findATender.FetchAll(findatender.RequestOptions{
+				Since:            since,
+				RateLimitRetries: rateLimitRetries,
+			})
 			if err != nil {
 				return err
 			}

internal/config/README.md

@@ -72,7 +72,9 @@ supported.
 | `sources.<source>.enabled` | bool | `false` | For every source |
 | `sources.<source>.interval` | duration | `10m` | For every source; e.g. `30s`, `1h`. Must be greater than 0; turn a source off with `enabled` |
 | `sources.find_a_tender.api_root` | string | `https://www.find-tender.service.gov.uk/api/1.0` | URL of the Find a Tender API. Empty: the source fails to start if enabled, and `fetcher fts-once` fails |
+| `sources.find_a_tender.rate_limit_retries` | int | `5` | How many times a request refused for the API's rate limit is waited out (for its `Retry-After`) and retried. 0 fails at the first; can't be negative. `fetcher fts-once --ignore-rate-limit` uses 0 |
 | `sources.fts_mock.api_root` | string | | URL of the mock FTS server (see `internal/mock`). Empty: the source fails to start if enabled |
+| `sources.fts_mock.rate_limit_retries` | int | `5` | As for `find_a_tender` |
 | `sources.sleeper.block_shutdown` | bool | `false` | Test source: if true, its sleep ignores shutdown |
 
 The current sources are `find_a_tender`, `fts_mock` and `sleeper`. See

internal/config/conf_sections_test.go

@@ -56,6 +56,14 @@ func TestDefaults(t *testing.T) {
 	if conf.Sources.FTSMock.APIRoot != "" {
 		t.Errorf("FTSMock.APIRoot = %q; want empty", conf.Sources.FTSMock.APIRoot)
 	}
+	if conf.Sources.FindATender.RateLimitRetries != 5 ||
+		conf.Sources.FTSMock.RateLimitRetries != 5 {
+		t.Errorf(
+			"RateLimitRetries = %d, %d; want 5, 5",
+			conf.Sources.FindATender.RateLimitRetries,
+			conf.Sources.FTSMock.RateLimitRetries,
+		)
+	}
 	for name, src := range map[string]JSCSourcesSource{
 		"find_a_tender": conf.Sources.FindATender.JSCSourcesSource,
 		"fts_mock":      conf.Sources.FTSMock.JSCSourcesSource,
@@ -236,6 +244,20 @@ func TestConcreteSourceSections(t *testing.T) {
 				}
 				wantMap["api_root"] = "http://localhost:8080"
 				wantFlat[tt.prefix+".api_root"] = "http://localhost:8080"
+
+				// The Find a Tender sources' own validate must still check
+				// interval, as JSCSourcesSource's does.
+				if err := s.SetValue("rate_limit_retries", "3"); err != nil {
+					t.Fatalf("SetValue(rate_limit_retries): %v", err)
+				}
+				if err := s.SetValue("rate_limit_retries", "-1"); err == nil {
+					t.Error("SetValue(rate_limit_retries, -1) succeeded")
+				}
+				if err := s.SetValue("interval", "0s"); err == nil {
+					t.Error("SetValue(interval, 0s) succeeded")
+				}
+				wantMap["rate_limit_retries"] = 3
+				wantFlat[tt.prefix+".rate_limit_retries"] = 3
 			} else if _, err := s.TypeOf("api_root"); err == nil {
 				t.Error("TypeOf(api_root) succeeded on a source without one")
 			}
@@ -295,8 +317,8 @@ func TestSourcesSection(t *testing.T) {
 	if flat["sources.fts_mock.api_root"] != "http://mock" {
 		t.Errorf("Flatten missing sources.fts_mock.api_root: %v", flat)
 	}
-	if len(flat) != 10 {
-		t.Errorf("Flatten has %d keys; want 10: %v", len(flat), flat)
+	if len(flat) != 12 {
+		t.Errorf("Flatten has %d keys; want 12: %v", len(flat), flat)
 	}
 
 	if err := c.SetValue("skip_rules_file", "rules.json"); err != nil ||

internal/config/conf_source_findatender.go

@@ -7,10 +7,13 @@ const JSCKSourceFindATender string = "find_a_tender"
 
 // JSCSourceFindATender configures fetching from the Find a Tender
 // OCDS API at APIRoot. Empty: the source fails to start if it is enabled,
-// and `fetcher fts-once` fails.
+// and `fetcher fts-once` fails. RateLimitRetries is how many times a request
+// refused for the API's rate limit is waited out and retried; 0 fails at the
+// first, and it can't be negative.
 type JSCSourceFindATender struct {
 	JSCSourcesSource
 	APIRoot          string `cfg:"api_root"           default:"https://www.find-tender.service.gov.uk/api/1.0"`
+	RateLimitRetries int    `cfg:"rate_limit_retries" default:"5"`
 }
 
 // ---------- Section methods (see Section) ----------
@@ -37,3 +40,12 @@ func (c JSCSourceFindATender) ValueOf(key string) (any, error) {
 func (c *JSCSourceFindATender) SetValue(key string, value any) error {
 	return setValue(c, key, value)
 }
+
+// ---------- Section specific methods ----------
+
+// validate checks values before they are stored (see validator), as
+// JSCSourcesSource's does, which this one hides. rate_limit_retries can't be
+// negative.
+func (c *JSCSourceFindATender) validate(key string, value any) error {
+	return validateRateLimitRetries(&c.JSCSourcesSource, key, value)
+}

internal/config/conf_source_ftsmock.go

@@ -7,10 +7,12 @@ const JSCKSourceFTSMock string = "fts_mock"
 
 // JSCSourceFTSMock configures fetching from a Find a Tender mock
 // server (cmd/mock-fts) at APIRoot, which has no default. Empty: the source
-// fails to start if it is enabled.
+// fails to start if it is enabled. RateLimitRetries is as for
+// JSCSourceFindATender.
 type JSCSourceFTSMock struct {
 	JSCSourcesSource
 	APIRoot          string `cfg:"api_root"`
+	RateLimitRetries int    `cfg:"rate_limit_retries" default:"5"`
 }
 
 // ---------- Section methods (see Section) ----------
@@ -37,3 +39,12 @@ func (c JSCSourceFTSMock) ValueOf(key string) (any, error) {
 func (c *JSCSourceFTSMock) SetValue(key string, value any) error {
 	return setValue(c, key, value)
 }
+
+// ---------- Section specific methods ----------
+
+// validate checks values before they are stored (see validator), as
+// JSCSourcesSource's does, which this one hides. rate_limit_retries can't be
+// negative.
+func (c *JSCSourceFTSMock) validate(key string, value any) error {
+	return validateRateLimitRetries(&c.JSCSourcesSource, key, value)
+}

internal/config/conf_sources.go

@@ -97,3 +97,17 @@ func (c *JSCSourcesSource) validate(key string, value any) error {
 	}
 	return nil
 }
+
+// validateRateLimitRetries is the validate of a Find a Tender source's
+// section: source's checks, then rate_limit_retries, which can't be
+// negative.
+func validateRateLimitRetries(
+	source *JSCSourcesSource, key string, value any,
+) error {
+	if key == "rate_limit_retries" && value.(int) < 0 {
+		return errors.New(
+			"config: rate_limit_retries can't be negative; 0 fails at the first rate limit",
+		)
+	}
+	return source.validate(key, value)
+}

internal/fetcher/backfill.go

@@ -0,0 +1,90 @@
+package fetcher
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"time"
+
+	"git.avsdev.uk/Go/job-search/internal/types"
+	"git.avsdev.uk/Go/job-search/internal/util"
+)
+
+// backfillChunk is the longest window a backfill fetches at once (the user's
+// choice: fixed, not a setting).
+const backfillChunk = 7 * 24 * time.Hour
+
+// fetchWindowFunc fetches the opportunities a source updated from since up
+// to until; a zero until means no bound. If it fails partway, it returns
+// what it fetched before the error along with it.
+type fetchWindowFunc func(
+	ctx context.Context, since, until time.Time,
+) ([]*types.Opportunity, error)
+
+// backfill catches a source up from since, before its worker starts, so a
+// long window (an early --since, or a long time offline) isn't fetched in
+// one go. It fetches backfillChunk windows, bounded on both sides, oldest
+// first, saving each before fetching the next, until less than a window is
+// left before now. It returns where the worker carries on from: since when
+// it is zero (the client picks) or within backfillChunk of now.
+//
+// A window that fails is saved as far as it got, and fetched again from its
+// start after interval, until it succeeds or ctx is cancelled. Cancelling
+// ctx returns the start of the window it stopped in.
+//
+// logger is given untagged, as to the workers (see Fetcher.Run); source is
+// the source's name, for the backfill's own lines.
+func backfill(
+	ctx context.Context,
+	logger *slog.Logger,
+	source string,
+	interval time.Duration,
+	fetch fetchWindowFunc,
+	save saveFunc,
+	since time.Time,
+) time.Time {
+	backfillLogger := tagged(logger).With("source", source)
+
+	for !since.IsZero() && time.Since(since) > backfillChunk {
+		until := since.Add(backfillChunk)
+		backfillLogger.Info("backfilling", "since", since, "until", until)
+
+		opportunities, err := fetch(ctx, since, until)
+		err = saveFetched(ctx, logger, save, opportunities, err)
+		switch {
+		case err == nil || errors.Is(err, util.ErrWorkerNoWork):
+			since = until
+		case ctx.Err() != nil:
+			return since
+		default:
+			backfillLogger.Error(
+				"backfill window failed, retrying",
+				"since", since,
+				"until", until,
+				"error", err,
+				"retry_in", interval,
+			)
+			if sleep(ctx, false, interval) != nil {
+				return since
+			}
+		}
+	}
+
+	return since
+}
+
+// saveFetched saves what a fetch returned, even when it failed partway, so
+// nothing fetched is lost. It returns fetchErr, joined with save's error if
+// that failed too.
+func saveFetched(
+	ctx context.Context,
+	logger *slog.Logger,
+	save saveFunc,
+	opportunities []*types.Opportunity,
+	fetchErr error,
+) error {
+	if len(opportunities) == 0 {
+		return fetchErr
+	}
+	return errors.Join(fetchErr, save(ctx, logger, opportunities))
+}

internal/fetcher/backfill_test.go

@@ -0,0 +1,293 @@
+package fetcher
+
+import (
+	"context"
+	"errors"
+	"io"
+	"log/slog"
+	"net/http"
+	"net/http/httptest"
+	"slices"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"git.avsdev.uk/Go/job-search/internal/config"
+	"git.avsdev.uk/Go/job-search/internal/types"
+)
+
+var errTestFetch = errors.New("test fetch failed")
+
+// span describes the window from since to until, for comparing windows.
+func span(since, until time.Time) string {
+	return since.Format(time.RFC3339) + "/" + until.Format(time.RFC3339)
+}
+
+// recorder is a fake source for backfill: it records the windows fetched
+// and the saves, in order, in calls.
+type recorder struct {
+	mu      sync.Mutex
+	calls   []string
+	windows []string
+	saved   int
+	// fail returns the error for the nth fetch (from 1), if any.
+	fail func(n int) error
+}
+
+func (r *recorder) fetch(
+	ctx context.Context, since, until time.Time,
+) ([]*types.Opportunity, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.windows = append(r.windows, span(since, until))
+	r.calls = append(r.calls, "fetch")
+	var err error
+	if r.fail != nil {
+		err = r.fail(len(r.windows))
+	}
+	opportunity := &types.Opportunity{
+		UUID: "ocds-1",
+	}
+	return []*types.Opportunity{opportunity}, err
+}
+
+func (r *recorder) save(
+	ctx context.Context, logger *slog.Logger, opportunities []*types.Opportunity,
+) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.saved += len(opportunities)
+	r.calls = append(r.calls, "save")
+	return nil
+}
+
+func TestBackfillWindows(t *testing.T) {
+	since := time.Now().Add(-20 * 24 * time.Hour)
+	r := &recorder{}
+
+	got := backfill(
+		context.Background(),
+		slog.Default(),
+		"Test",
+		time.Millisecond,
+		r.fetch,
+		r.save,
+		since,
+	)
+
+	second := since.Add(backfillChunk)
+	third := since.Add(2 * backfillChunk)
+	want := []string{span(since, second), span(second, third)}
+	if !slices.Equal(r.windows, want) {
+		t.Errorf("windows = %v; want %v", r.windows, want)
+	}
+	if !got.Equal(third) {
+		t.Errorf("backfill = %v; want the end of the last window", got)
+	}
+	// Each window is saved before the next is fetched.
+	if calls := strings.Join(r.calls, ","); calls != "fetch,save,fetch,save" {
+		t.Errorf("calls = %s", calls)
+	}
+}
+
+// Nothing is backfilled for a zero since (the client picks) or one within a
+// window of now: the worker fetches it.
+func TestBackfillNothingToDo(t *testing.T) {
+	for name, since := range map[string]time.Time{
+		"zero":   {},
+		"recent": time.Now().Add(-3 * 24 * time.Hour),
+		"future": time.Now().Add(time.Hour),
+	} {
+		t.Run(name, func(t *testing.T) {
+			r := &recorder{}
+			got := backfill(
+				context.Background(),
+				slog.Default(),
+				"Test",
+				time.Millisecond,
+				r.fetch,
+				r.save,
+				since,
+			)
+			if !got.Equal(since) || len(r.windows) != 0 {
+				t.Errorf(
+					"backfill = %v after %d fetches; want %v and none",
+					got, len(r.windows), since,
+				)
+			}
+		})
+	}
+}
+
+// A failed window is saved as far as it got and fetched again from its
+// start; finding nothing moves on like a success.
+func TestBackfillRetriesFailedWindow(t *testing.T) {
+	since := time.Now().Add(-15 * 24 * time.Hour)
+	r := &recorder{
+		fail: func(n int) error {
+			switch n {
+			case 1, 2:
+				return errTestFetch
+			case 3:
+				return ErrNoNewReleases
+			default:
+				return nil
+			}
+		},
+	}
+
+	got := backfill(
+		context.Background(),
+		slog.Default(),
+		"Test",
+		time.Millisecond,
+		r.fetch,
+		r.save,
+		since,
+	)
+
+	second := since.Add(backfillChunk)
+	third := since.Add(2 * backfillChunk)
+	first := span(since, second)
+	want := []string{first, first, first, span(second, third)}
+	if !slices.Equal(r.windows, want) {
+		t.Errorf("windows = %v; want %v", r.windows, want)
+	}
+	if r.saved != 4 {
+		t.Errorf("saved %d opportunities; want 4 (failed windows too)", r.saved)
+	}
+	if !got.Equal(third) {
+		t.Errorf("backfill = %v; want %v", got, third)
+	}
+}
+
+// Cancelling stops the backfill during its wait, returning the start of the
+// window it stopped in.
+func TestBackfillCancelled(t *testing.T) {
+	since := time.Now().Add(-15 * 24 * time.Hour)
+	ctx, cancel := context.WithCancel(context.Background())
+	r := &recorder{
+		fail: func(n int) error {
+			cancel()
+			return errTestFetch
+		},
+	}
+
+	start := time.Now()
+	got := backfill(
+		ctx, slog.Default(), "Test", time.Hour, r.fetch, r.save, since,
+	)
+
+	if time.Since(start) > 5*time.Second {
+		t.Error("cancelling did not stop the wait")
+	}
+	if !got.Equal(since) || len(r.windows) != 1 {
+		t.Errorf(
+			"backfill = %v after %d fetches; want %v after 1",
+			got, len(r.windows), since,
+		)
+	}
+}
+
+func TestSaveFetched(t *testing.T) {
+	r := &recorder{}
+	partial := []*types.Opportunity{
+		{
+			UUID: "ocds-1",
+		},
+	}
+
+	err := saveFetched(
+		context.Background(), slog.Default(), r.save, partial, errTestFetch,
+	)
+	if !errors.Is(err, errTestFetch) || r.saved != 1 {
+		t.Errorf(
+			"partial: err = %v, saved %d; want the fetch error, 1", err, r.saved,
+		)
+	}
+
+	err = saveFetched(
+		context.Background(), slog.Default(), r.save, nil, ErrNoNewReleases,
+	)
+	if !errors.Is(err, ErrNoNewReleases) || r.saved != 1 {
+		t.Errorf(
+			"nothing: err = %v, saved %d; want ErrNoNewReleases, 1", err, r.saved,
+		)
+	}
+
+	failing := func(context.Context, *slog.Logger, []*types.Opportunity) error {
+		return errTestFetch
+	}
+	err = saveFetched(context.Background(), slog.Default(), failing, partial, nil)
+	if !errors.Is(err, errTestFetch) {
+		t.Errorf("failed save: err = %v; want the save's error", err)
+	}
+}
+
+// Run backfills each Find a Tender source in windows bounded on both sides,
+// then its worker fetches the rest with no upper bound.
+func TestRunBackfills(t *testing.T) {
+	for _, source := range []string{"FindATender", "FTSMock"} {
+		t.Run(source, func(t *testing.T) {
+			var (
+				mu    sync.Mutex
+				until []string
+			)
+			srv := httptest.NewServer(
+				http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+					mu.Lock()
+					until = append(until, r.URL.Query().Get("updatedTo"))
+					mu.Unlock()
+					io.WriteString(w, emptyPackage)
+				}),
+			)
+			t.Cleanup(srv.Close)
+
+			sources := &config.JSCSources{}
+			if source == "FindATender" {
+				sources.FindATender.Enabled = true
+				sources.FindATender.Interval = time.Hour
+				sources.FindATender.APIRoot = srv.URL
+			} else {
+				sources.FTSMock.Enabled = true
+				sources.FTSMock.Interval = time.Hour
+				sources.FTSMock.APIRoot = srv.URL
+			}
+
+			since := time.Now().UTC().Add(-15 * 24 * time.Hour)
+			ctx, cancel := context.WithCancel(context.Background())
+			done := make(chan error, 1)
+			go func() {
+				done <- New(sources).Run(ctx, since)
+			}()
+
+			deadline := time.Now().Add(5 * time.Second)
+			for {
+				mu.Lock()
+				n := len(until)
+				mu.Unlock()
+				if n >= 3 || time.Now().After(deadline) {
+					break
+				}
+				time.Sleep(5 * time.Millisecond)
+			}
+			cancel()
+			if err := <-done; err != nil {
+				t.Fatalf("Run = %v", err)
+			}
+
+			const layout = "2006-01-02T15:04:05"
+			want := []string{
+				since.Add(backfillChunk).Format(layout),
+				since.Add(2 * backfillChunk).Format(layout),
+				"",
+			}
+			mu.Lock()
+			defer mu.Unlock()
+			if !slices.Equal(until, want) {
+				t.Errorf("updatedTo of each request = %q; want %q", until, want)
+			}
+		})
+	}
+}

internal/fetcher/fetcher_test.go

@@ -77,7 +77,7 @@ var fetchers = map[string]fetchFunc{
 		conf := &config.JSCSourceFindATender{
 			APIRoot: apiRoot,
 		}
-		return fetchFindATender(ctx, slog.Default(), conf, since)
+		return fetchFindATender(ctx, slog.Default(), conf, since, time.Time{})
 	},
 	"FTSMock": func(
 		ctx context.Context, apiRoot string, since time.Time,
@@ -85,7 +85,7 @@ var fetchers = map[string]fetchFunc{
 		conf := &config.JSCSourceFTSMock{
 			APIRoot: apiRoot,
 		}
-		return fetchFTSMock(ctx, slog.Default(), conf, since)
+		return fetchFTSMock(ctx, slog.Default(), conf, since, time.Time{})
 	},
 }
 

internal/fetcher/worker_findatender.go

@@ -17,7 +17,8 @@ import (
 const findATenderSource = "FindATender"
 
 // runFindATender polls Find a Tender every configured interval until ctx
-// is cancelled, passing what each run fetches to save.
+// is cancelled, passing what each run fetches to save, even when the run
+// fails partway. A long window is backfilled first (see backfill).
 func runFindATender(
 	ctx context.Context,
 	logger *slog.Logger,
@@ -29,29 +30,43 @@ func runFindATender(
 		return errors.New("fetcher: sources.find_a_tender.api_root is not set")
 	}
 
+	fetch := func(
+		ctx context.Context, since, until time.Time,
+	) ([]*types.Opportunity, error) {
+		return fetchFindATender(ctx, logger, conf, since, until)
+	}
+
+	since = backfill(
+		ctx,
+		logger,
+		findATenderSource,
+		conf.Interval,
+		fetch,
+		save,
+		since,
+	)
+
 	worker := util.Worker{
 		Logger:   tagged(logger),
 		Name:     findATenderSource,
 		Interval: conf.Interval,
 		Task: func(ctx context.Context, since time.Time) error {
-			opportunities, err := fetchFindATender(ctx, logger, conf, since)
-			if err != nil {
-				return err
-			}
-			return save(ctx, logger, opportunities)
+			opportunities, err := fetch(ctx, since, time.Time{})
+			return saveFetched(ctx, logger, save, opportunities, err)
 		},
 	}
 	return worker.Run(ctx, since)
 }
 
-// fetchFindATender fetches everything updated since since from Find a
-// Tender. Finding nothing returns ErrNoNewReleases, so the worker still moves
-// its checkpoint on.
+// fetchFindATender fetches everything updated from since up to until (zero:
+// no bound) from Find a Tender. If it fails partway, what it fetched before
+// the error is returned along with it. Finding nothing returns
+// ErrNoNewReleases, so the worker still moves its checkpoint on.
 func fetchFindATender(
 	ctx context.Context,
 	logger *slog.Logger,
 	conf *config.JSCSourceFindATender,
-	since time.Time,
+	since, until time.Time,
 ) ([]*types.Opportunity, error) {
 	opportunities, err := util.AwaitTask(
 		ctx,
@@ -60,17 +75,18 @@ func fetchFindATender(
 				conf.APIRoot, findatender.WithLogger(logger),
 			)
 			defer findATender.Shutdown()
-			return findATender.With(ctx).FetchAll(since, true)
+			return findATender.With(ctx).FetchAll(findatender.RequestOptions{
+				Since:            since,
+				Until:            until,
+				RateLimitRetries: conf.RateLimitRetries,
+			})
 		},
 	)
 	if errors.Is(err, findatender.ErrNoReleases) {
 		return nil, ErrNoNewReleases
 	}
-	if err != nil {
-		return nil, err
-	}
 
 	tagged(logger).Info("fetched opportunities", "count", len(opportunities))
 
-	return opportunities, nil
+	return opportunities, err
 }

internal/fetcher/worker_ftsmock.go

@@ -16,7 +16,7 @@ import (
 const ftsMockSource = "FTSMock"
 
 // runFTSMock polls the mock Find a Tender server every configured interval
-// until ctx is cancelled, passing what each run fetches to save.
+// until ctx is cancelled, as runFindATender polls Find a Tender.
 func runFTSMock(
 	ctx context.Context,
 	logger *slog.Logger,
@@ -28,31 +28,44 @@ func runFTSMock(
 		return errors.New("fetcher: sources.fts_mock.api_root is not set")
 	}
 
+	fetch := func(
+		ctx context.Context, since, until time.Time,
+	) ([]*types.Opportunity, error) {
+		return fetchFTSMock(ctx, logger, conf, since, until)
+	}
+
+	since = backfill(
+		ctx,
+		logger,
+		ftsMockSource,
+		conf.Interval,
+		fetch,
+		save,
+		since,
+	)
+
 	worker := util.Worker{
 		Logger:   tagged(logger),
 		Name:     ftsMockSource,
 		Interval: conf.Interval,
 		Task: func(ctx context.Context, since time.Time) error {
-			opportunities, err := fetchFTSMock(ctx, logger, conf, since)
-			if err != nil {
-				return err
-			}
-			return save(ctx, logger, opportunities)
+			opportunities, err := fetch(ctx, since, time.Time{})
+			return saveFetched(ctx, logger, save, opportunities, err)
 		},
 	}
 	return worker.Run(ctx, since)
 }
 
-// fetchFTSMock fetches everything updated since since from the mock Find a
-// Tender server. The mock serves the real API's format, so the converter
-// marks the opportunities as Find a Tender's; their Source is set to
-// ftsMockSource here instead. Finding nothing returns ErrNoNewReleases, so
-// the worker still moves its checkpoint on.
+// fetchFTSMock fetches everything updated from since up to until (zero: no
+// bound) from the mock Find a Tender server, as fetchFindATender does. The
+// mock serves the real API's format, so the converter marks the
+// opportunities as Find a Tender's; their Source is set to ftsMockSource here
+// instead.
 func fetchFTSMock(
 	ctx context.Context,
 	logger *slog.Logger,
 	conf *config.JSCSourceFTSMock,
-	since time.Time,
+	since, until time.Time,
 ) ([]*types.Opportunity, error) {
 	opportunities, err := util.AwaitTask(
 		ctx,
@@ -61,15 +74,16 @@ func fetchFTSMock(
 				conf.APIRoot, findatender.WithLogger(logger),
 			)
 			defer findATender.Shutdown()
-			return findATender.With(ctx).FetchAll(since, true)
+			return findATender.With(ctx).FetchAll(findatender.RequestOptions{
+				Since:            since,
+				Until:            until,
+				RateLimitRetries: conf.RateLimitRetries,
+			})
 		},
 	)
 	if errors.Is(err, findatender.ErrNoReleases) {
 		return nil, ErrNoNewReleases
 	}
-	if err != nil {
-		return nil, err
-	}
 
 	for _, opportunity := range opportunities {
 		opportunity.Source = ftsMockSource
@@ -77,5 +91,5 @@ func fetchFTSMock(
 
 	tagged(logger).Info("fetched opportunities", "count", len(opportunities))
 
-	return opportunities, nil
+	return opportunities, err
 }

internal/findatender/api.go

@@ -20,6 +20,10 @@ import (
 	"git.avsdev.uk/Go/job-search/internal/logging"
 )
 
+// apiTimeLayout is the layout of updatedFrom and updatedTo: wall-clock time
+// with no offset, which the API reads as UK time.
+const apiTimeLayout = "2006-01-02T15:04:05"
+
 // Errors returned when fetching from the API.
 var (
 	// ErrNoNext means a release package has no next page link.
@@ -27,7 +31,8 @@ var (
 	// ErrNoReleases means the API returned no releases.
 	ErrNoReleases = errors.New("findatender: no releases")
 	// ErrRateLimitExceeded means the API refused a request with 429 Too Many
-	// Requests, and waiting for it was not allowed or did not help.
+	// Requests, and waiting for it was not allowed or did not help (see
+	// RequestOptions.RateLimitRetries).
 	ErrRateLimitExceeded = errors.New("findatender: rate limit has been exceeded")
 	// ErrFetchCancelled means the fetch's context was cancelled while waiting
 	// out a rate limit.
@@ -40,11 +45,15 @@ var (
 type RequestOptions struct {
 	// Since asks for releases updated from this time. Zero means 24 hours ago.
 	Since time.Time
+	// Until asks for releases updated up to this time. Zero means no bound
+	// (the API's: now).
+	Until time.Time
 	// Limit is the number of releases per page. Zero means 100.
 	Limit int
-	// AllowRateLimit waits out a 429 response and retries once, instead of
-	// failing with ErrRateLimitExceeded.
-	AllowRateLimit bool
+	// RateLimitRetries is how many times a request refused with 429 Too Many
+	// Requests is retried, each after waiting as the response's Retry-After
+	// asks, before failing with ErrRateLimitExceeded. 0 fails at the first.
+	RateLimitRetries int
 }
 
 // ---------- Utility methods for handling API irregularities ----------
@@ -409,55 +418,48 @@ func logTrace(logger *slog.Logger, res *resty.Response) {
 	}
 }
 
-// fetchURI performs a fetch against the API. If the API returns with a rate
-// limit reached response and allowRateLimit is true, then wait and retry like a
-// good client. If allowRateLimit is false, return to the caller informing them
-// of the rate limit exceeded error. If, after waiting, the API still thinks we
-// have exceeded the rate limit, bail out anyway.
+// fetchURI performs a fetch against the API. A response of 429 Too Many
+// Requests is waited out for as long as its Retry-After asks and the request
+// retried, up to rateLimitRetries times, like a good client; one more fails
+// with ErrRateLimitExceeded.
 func (ft *FindATender) fetchURI(
-	uri string, allowRateLimit bool,
+	uri string, rateLimitRetries int,
 ) (*resty.Response, error) {
-	res, err := fetchURI(ft, uri)
-	if err != nil {
-		return nil, err
-	}
-
 	ctx := ft.ctx
 	if ctx == nil {
 		ctx = context.Background()
 	}
 
-	if res.StatusCode() == 429 {
+	res, err := fetchURI(ft, uri)
+	for retry := 1; err == nil && res.StatusCode() == 429; retry++ {
 		closeBody(res)
 
-		if !allowRateLimit {
+		if retry > rateLimitRetries {
 			return nil, ErrRateLimitExceeded
 		}
 
 		retryAfterText := res.Header().Get("Retry-After")
-		retryAfterSeconds, err := strconv.ParseInt(retryAfterText, 10, 32)
-		if err != nil {
-			return nil, fmt.Errorf("unknown retry-after value: %w", err)
+		retryAfterSeconds, parseErr := strconv.ParseInt(retryAfterText, 10, 32)
+		if parseErr != nil {
+			return nil, fmt.Errorf("unknown retry-after value: %w", parseErr)
 		}
 
 		retryAfter := time.Duration(retryAfterSeconds) * time.Second
 		ft.logger().Warn(
 			"api rate limit hit, waiting as asked",
 			"retry_after", retryAfter,
+			"retry", retry,
+			"retries", rateLimitRetries,
 		)
 		if rateSleep(ctx, retryAfter) != nil {
 			return nil, ErrFetchCancelled
 		}
 
 		res, err = fetchURI(ft, uri)
+	}
 	if err != nil {
 		return nil, err
 	}
-		if res.StatusCode() == 429 {
-			closeBody(res)
-			return nil, ErrRateLimitExceeded
-		}
-	}
 
 	if res.StatusCode() != 200 {
 		closeBody(res)
@@ -482,23 +484,26 @@ func (ft *FindATender) fetchWithOpts(
 	uri := fmt.Sprintf(
 		"%s/ocdsReleasePackages?updatedFrom=%s&limit=%d",
 		ft.apiRoot,
-		opts.Since.Format("2006-01-02T15:04:05"),
+		opts.Since.Format(apiTimeLayout),
 		opts.Limit,
 	)
+	if !opts.Until.IsZero() {
+		uri += "&updatedTo=" + opts.Until.Format(apiTimeLayout)
+	}
 
-	return ft.fetchURI(uri, opts.AllowRateLimit)
+	return ft.fetchURI(uri, opts.RateLimitRetries)
 }
 
 // fetchNext fetches the page pkg's "Next" link points to. It returns
 // ErrNoNext if there is no "Next" link; the caller can decide whether asking
 // for it was an error.
 func (ft *FindATender) fetchNext(
-	releasePackage *ocds.ReleasePackage, allowRateLimit bool,
+	releasePackage *ocds.ReleasePackage, rateLimitRetries int,
 ) (*resty.Response, error) {
 	if releasePackage.Links == nil || releasePackage.Links.Next == "" {
 		return nil, ErrNoNext
 	}
-	return ft.fetchURI(releasePackage.Links.Next, allowRateLimit)
+	return ft.fetchURI(releasePackage.Links.Next, rateLimitRetries)
 }
 
 // fetchOnce makes one call on the API. It does not follow any "Next" links.
@@ -521,7 +526,9 @@ func (ft *FindATender) fetchOnce(opts RequestOptions) ([]ocds.Release, error) {
 }
 
 // fetchAll makes a call on the API. If there are any "Next" links, it calls
-// them in succession until the end has been reached.
+// them in succession until the end has been reached. If a page fails, the
+// releases of the pages before it are returned with the error, so the caller
+// can keep them.
 func (ft *FindATender) fetchAll(opts RequestOptions) ([]ocds.Release, error) {
 	var releases []ocds.Release
 
@@ -551,17 +558,17 @@ func (ft *FindATender) fetchAll(opts RequestOptions) ([]ocds.Release, error) {
 			seen[releasePackage.Links.Next] = true
 		}
 
-		res, err := ft.fetchNext(releasePackage, opts.AllowRateLimit)
+		res, err := ft.fetchNext(releasePackage, opts.RateLimitRetries)
 		if errors.Is(err, ErrNoNext) {
 			break
 		}
 		if err != nil {
-			return nil, err
+			return releases, err
 		}
 
 		releasePackage, err = safeUnpackReleases(ft.logger(), res)
 		if err != nil {
-			return nil, err
+			return releases, err
 		}
 		releases = append(releases, releasePackage.Releases...)
 	}

internal/findatender/api_test.go

@@ -129,6 +129,20 @@ func TestFetchWithOptsBuildsURI(t *testing.T) {
 	if !strings.HasSuffix(gotQuery, "&limit=100") {
 		t.Errorf("default limit not applied: %q", gotQuery)
 	}
+
+	res, err = ft.fetchWithOpts(RequestOptions{
+		Since: since,
+		Until: since.Add(7 * 24 * time.Hour),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	res.Body.Close()
+	want := "updatedFrom=2026-10-08T09:30:15&limit=100" +
+		"&updatedTo=2026-10-15T09:30:15"
+	if gotQuery != want {
+		t.Errorf("query = %q, want %q", gotQuery, want)
+	}
 }
 
 // A zero Since defaults to 24 hours ago.
@@ -160,29 +174,15 @@ func TestFetchURIRateLimit(t *testing.T) {
 	tests := []struct {
 		name        string
 		retryAfter  string
-		allowRateLimit bool
+		retries     int
 		wantErr     error
 		wantErrText string
 		wantHits    int32
 	}{
-		{"not allowed", "0", false, ErrRateLimitExceeded, "", 1},
-		{"allowed, retries", "0", true, nil, "", 2},
-		{
-			"allowed, bad retry-after",
-			"soon",
-			true,
-			nil,
-			"unknown retry-after value",
-			1,
-		},
-		{
-			"allowed, missing retry-after",
-			"",
-			true,
-			nil,
-			"unknown retry-after value",
-			1,
-		},
+		{"no retries", "0", 0, ErrRateLimitExceeded, "", 1},
+		{"retries", "0", 1, nil, "", 2},
+		{"bad retry-after", "soon", 1, nil, "unknown retry-after value", 1},
+		{"missing retry-after", "", 1, nil, "unknown retry-after value", 1},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -202,7 +202,7 @@ func TestFetchURIRateLimit(t *testing.T) {
 				},
 			)
 
-			res, err := ft.fetchURI(srv.URL+"/x", tt.allowRateLimit)
+			res, err := ft.fetchURI(srv.URL+"/x", tt.retries)
 			switch {
 			case tt.wantErr != nil:
 				if !errors.Is(err, tt.wantErr) {
@@ -241,7 +241,7 @@ func TestFetchURIRateLimitCancelled(t *testing.T) {
 	}()
 
 	start := time.Now()
-	_, err := withCtx.fetchURI(srv.URL, true)
+	_, err := withCtx.fetchURI(srv.URL, 1)
 	if !errors.Is(err, ErrFetchCancelled) {
 		t.Fatalf("err = %v, want ErrFetchCancelled", err)
 	}
@@ -250,20 +250,47 @@ func TestFetchURIRateLimitCancelled(t *testing.T) {
 	}
 }
 
-func TestFetchURIRateLimitTwice(t *testing.T) {
-	// A second 429 after waiting is reported as ErrRateLimitExceeded.
+// Each 429 is waited out until the retries run out; the next one is
+// reported as ErrRateLimitExceeded.
+func TestFetchURIRateLimitRepeated(t *testing.T) {
+	tests := []struct {
+		name     string
+		limited  int32 // requests answered 429 before one succeeds
+		retries  int
+		wantErr  error
+		wantHits int32
+	}{
+		{"fewer than the retries", 3, 5, nil, 4},
+		{"as many as the retries", 3, 3, nil, 4},
+		{"more than the retries", 3, 2, ErrRateLimitExceeded, 3},
+		{"never succeeds", 1000, 4, ErrRateLimitExceeded, 5},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
 			var hits atomic.Int32
-	ft, srv := newTestFT(t, false, func(w http.ResponseWriter, r *http.Request) {
-		hits.Add(1)
+			ft, srv := newTestFT(
+				t,
+				false,
+				func(w http.ResponseWriter, r *http.Request) {
+					if hits.Add(1) <= tt.limited {
 						w.Header().Set("Retry-After", "0")
 						w.WriteHeader(http.StatusTooManyRequests)
-	})
-	_, err := ft.fetchURI(srv.URL, true)
-	if !errors.Is(err, ErrRateLimitExceeded) {
-		t.Fatalf("err = %v, want ErrRateLimitExceeded", err)
+						return
 					}
-	if n := hits.Load(); n != 2 {
-		t.Errorf("requests = %d, want 2", n)
+					io.WriteString(w, packageJSON("", releaseJSON("ocds-1", "1")))
+				},
+			)
+			res, err := ft.fetchURI(srv.URL, tt.retries)
+			if !errors.Is(err, tt.wantErr) {
+				t.Fatalf("err = %v, want %v", err, tt.wantErr)
+			}
+			if err == nil {
+				res.Body.Close()
+			}
+			if n := hits.Load(); n != tt.wantHits {
+				t.Errorf("requests = %d, want %d", n, tt.wantHits)
+			}
+		})
 	}
 }
 
@@ -281,7 +308,7 @@ func TestFetchURIUnexpectedStatus(t *testing.T) {
 					io.WriteString(w, packageJSON("", releaseJSON("ocds-1", "1")))
 				},
 			)
-			_, err := ft.fetchURI(srv.URL, true)
+			_, err := ft.fetchURI(srv.URL, 1)
 			if !errors.Is(err, ErrUnexpectedStatus) {
 				t.Fatalf("err = %v, want ErrUnexpectedStatus", err)
 			}
@@ -303,7 +330,7 @@ func TestFetchURIRateLimitThenError(t *testing.T) {
 		}
 		w.WriteHeader(http.StatusServiceUnavailable)
 	})
-	_, err := ft.fetchURI(srv.URL, true)
+	_, err := ft.fetchURI(srv.URL, 1)
 	if !errors.Is(err, ErrUnexpectedStatus) {
 		t.Fatalf("err = %v, want ErrUnexpectedStatus", err)
 	}
@@ -314,7 +341,7 @@ func TestFetchURIConnectionError(t *testing.T) {
 	})
 	srv.Close()
 	ft.httpClient.SetRetryCount(0)
-	if _, err := ft.fetchURI(srv.URL, true); err == nil {
+	if _, err := ft.fetchURI(srv.URL, 1); err == nil {
 		t.Fatal("expected an error from a closed server")
 	}
 }
@@ -325,7 +352,7 @@ func TestFetchURICancelledContext(t *testing.T) {
 	})
 	ctx, cancel := context.WithCancel(context.Background())
 	cancel()
-	if _, err := ft.With(ctx).fetchURI(srv.URL, true); err == nil {
+	if _, err := ft.With(ctx).fetchURI(srv.URL, 1); err == nil {
 		t.Fatal("expected an error from a cancelled context")
 	}
 }
@@ -404,7 +431,7 @@ func TestFetchNext(t *testing.T) {
 			Prev: srv.URL,
 		}},
 	} {
-		if _, err := ft.fetchNext(pkg, true); !errors.Is(err, ErrNoNext) {
+		if _, err := ft.fetchNext(pkg, 1); !errors.Is(err, ErrNoNext) {
 			t.Errorf("%s: err = %v, want ErrNoNext", name, err)
 		}
 	}
@@ -415,7 +442,7 @@ func TestFetchNext(t *testing.T) {
 				Next: srv.URL + "/page2",
 			},
 		},
-		true,
+		1,
 	)
 	if err != nil {
 		t.Fatal(err)
@@ -432,7 +459,7 @@ func responseFor(t *testing.T, body string) *resty.Response {
 	ft, srv := newTestFT(t, false, func(w http.ResponseWriter, r *http.Request) {
 		io.WriteString(w, body)
 	})
-	res, err := ft.fetchURI(srv.URL, false)
+	res, err := ft.fetchURI(srv.URL, 0)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -676,7 +703,8 @@ func TestFetchAllErrors(t *testing.T) {
 		t.Error("first page bad body: expected an error")
 	}
 
-	// A later page fails, either to fetch or to decode.
+	// A later page fails, either to fetch or to decode: the error is returned
+	// with the releases fetched before it.
 	for name, fail := range map[string]http.HandlerFunc{
 		"rate limited": func(w http.ResponseWriter, r *http.Request) {
 			w.WriteHeader(http.StatusTooManyRequests)
@@ -701,11 +729,16 @@ func TestFetchAllErrors(t *testing.T) {
 				},
 			)
 			srvURL = srv.URL
-			if _, err := ft.fetchAll(RequestOptions{
+			releases, err := ft.fetchAll(RequestOptions{
 				Since: time.Now(),
-			}); err == nil {
+			})
+			if err == nil {
 				t.Error("expected an error from the second page")
 			}
+			// The first page's releases are kept.
+			if len(releases) != 1 || releases[0].OCID != "ocds-1" {
+				t.Errorf("releases = %+v, want the first page's", releases)
+			}
 		})
 	}
 }

internal/findatender/findatender.go

@@ -76,34 +76,25 @@ func (ft *FindATender) Shutdown() {
 	}
 }
 
-// FetchOnce fetches the first page of releases updated since since, and
-// converts them to opportunities. If allowRateLimit is true, a 429 response
-// is waited out and retried; otherwise it fails with ErrRateLimitExceeded.
+// FetchOnce fetches the first page of the releases opts selects, and
+// converts them to opportunities.
 func (ft *FindATender) FetchOnce(
-	since time.Time, allowRateLimit bool,
+	opts RequestOptions,
 ) ([]*types.Opportunity, error) {
-	releases, err := ft.fetchOnce(RequestOptions{
-		Since:          since,
-		AllowRateLimit: allowRateLimit,
-	})
+	releases, err := ft.fetchOnce(opts)
 	if err != nil {
 		return nil, err
 	}
 	return Convert(releases), nil
 }
 
-// FetchAll fetches every page of releases updated since since, following
-// each page's next link, and converts them to opportunities. allowRateLimit
-// behaves as for FetchOnce.
+// FetchAll fetches every page of the releases opts selects, following each
+// page's next link, and converts them to opportunities. If a page fails, the
+// opportunities from the pages before it are returned with the error, so the
+// caller can keep them.
 func (ft *FindATender) FetchAll(
-	since time.Time, allowRateLimit bool,
+	opts RequestOptions,
 ) ([]*types.Opportunity, error) {
-	releases, err := ft.fetchAll(RequestOptions{
-		Since:          since,
-		AllowRateLimit: allowRateLimit,
-	})
-	if err != nil {
-		return nil, err
-	}
-	return Convert(releases), nil
+	releases, err := ft.fetchAll(opts)
+	return Convert(releases), err
 }

internal/findatender/findatender_test.go

@@ -82,7 +82,9 @@ func TestFetchOncePublic(t *testing.T) {
 		},
 		[]string{releaseJSON("ocds-3", "1")},
 	)
-	ops, err := ft.FetchOnce(time.Now(), false)
+	ops, err := ft.FetchOnce(RequestOptions{
+		Since: time.Now(),
+	})
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -91,7 +93,9 @@ func TestFetchOncePublic(t *testing.T) {
 	}
 
 	ft, _ = pagedServer(t, []string{})
-	if _, err := ft.FetchOnce(time.Now(), false); !errors.Is(err, ErrNoReleases) {
+	if _, err := ft.FetchOnce(RequestOptions{
+		Since: time.Now(),
+	}); !errors.Is(err, ErrNoReleases) {
 		t.Errorf("err = %v, want ErrNoReleases", err)
 	}
 }
@@ -102,7 +106,9 @@ func TestFetchAllPublic(t *testing.T) {
 		[]string{releaseJSON("ocds-1", "1"), releaseJSON("ocds-2", "1")},
 		[]string{releaseJSON("ocds-1", "2"), releaseJSON("ocds-3", "1")},
 	)
-	ops, err := ft.FetchAll(time.Now(), false)
+	ops, err := ft.FetchAll(RequestOptions{
+		Since: time.Now(),
+	})
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -120,7 +126,9 @@ func TestFetchAllPublic(t *testing.T) {
 	ft, _ = newTestFT(t, false, func(w http.ResponseWriter, r *http.Request) {
 		io.WriteString(w, "not json")
 	})
-	if _, err := ft.FetchAll(time.Now(), false); err == nil {
+	if _, err := ft.FetchAll(RequestOptions{
+		Since: time.Now(),
+	}); err == nil {
 		t.Error("expected an error")
 	}
 }