diff --git a/README.md b/README.md index 3704626..b6dbb8c 100644 --- a/README.md +++ b/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..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. diff --git a/internal/cli/fetch/fts_once.go b/internal/cli/fetch/fts_once.go index 47e24ba..58c7b83 100644 --- a/internal/cli/fetch/fts_once.go +++ b/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 } diff --git a/internal/config/README.md b/internal/config/README.md index 5c95c57..f537920 100644 --- a/internal/config/README.md +++ b/internal/config/README.md @@ -72,7 +72,9 @@ supported. | `sources..enabled` | bool | `false` | For every source | | `sources..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 diff --git a/internal/config/conf_sections_test.go b/internal/config/conf_sections_test.go index 37060e1..da2c88a 100644 --- a/internal/config/conf_sections_test.go +++ b/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 || diff --git a/internal/config/conf_source_findatender.go b/internal/config/conf_source_findatender.go index f5dcc35..f0c1c58 100644 --- a/internal/config/conf_source_findatender.go +++ b/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"` + 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) +} diff --git a/internal/config/conf_source_ftsmock.go b/internal/config/conf_source_ftsmock.go index 8945c07..1026230 100644 --- a/internal/config/conf_source_ftsmock.go +++ b/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"` + 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) +} diff --git a/internal/config/conf_sources.go b/internal/config/conf_sources.go index ba6a80b..1b917ec 100644 --- a/internal/config/conf_sources.go +++ b/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) +} diff --git a/internal/fetcher/backfill.go b/internal/fetcher/backfill.go new file mode 100644 index 0000000..73e2bde --- /dev/null +++ b/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)) +} diff --git a/internal/fetcher/backfill_test.go b/internal/fetcher/backfill_test.go new file mode 100644 index 0000000..5569721 --- /dev/null +++ b/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) + } + }) + } +} diff --git a/internal/fetcher/fetcher_test.go b/internal/fetcher/fetcher_test.go index f8916b4..7b710dc 100644 --- a/internal/fetcher/fetcher_test.go +++ b/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{}) }, } diff --git a/internal/fetcher/worker_findatender.go b/internal/fetcher/worker_findatender.go index 6421720..82d7c4a 100644 --- a/internal/fetcher/worker_findatender.go +++ b/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 } diff --git a/internal/fetcher/worker_ftsmock.go b/internal/fetcher/worker_ftsmock.go index f3bdb06..f9e147a 100644 --- a/internal/fetcher/worker_ftsmock.go +++ b/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 } diff --git a/internal/findatender/api.go b/internal/findatender/api.go index a50a48f..6bf09c2 100644 --- a/internal/findatender/api.go +++ b/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,54 +418,47 @@ 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 err != nil { + return nil, err } if res.StatusCode() != 200 { @@ -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...) } diff --git a/internal/findatender/api_test.go b/internal/findatender/api_test.go index 465e64e..91e39c8 100644 --- a/internal/findatender/api_test.go +++ b/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. @@ -158,31 +172,17 @@ func TestFetchWithOptsDefaultSince(t *testing.T) { func TestFetchURIRateLimit(t *testing.T) { tests := []struct { - name string - retryAfter string - allowRateLimit bool - wantErr error - wantErrText string - wantHits int32 + name string + retryAfter string + 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. - var hits atomic.Int32 - ft, srv := newTestFT(t, false, func(w http.ResponseWriter, r *http.Request) { - hits.Add(1) - 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) +// 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}, } - if n := hits.Load(); n != 2 { - t.Errorf("requests = %d, want 2", n) + 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) { + if hits.Add(1) <= tt.limited { + w.Header().Set("Retry-After", "0") + w.WriteHeader(http.StatusTooManyRequests) + return + } + 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) + } }) } } diff --git a/internal/findatender/findatender.go b/internal/findatender/findatender.go index 60cfc26..aeb28bc 100644 --- a/internal/findatender/findatender.go +++ b/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 } diff --git a/internal/findatender/findatender_test.go b/internal/findatender/findatender_test.go index eb08b6c..f3b1034 100644 --- a/internal/findatender/findatender_test.go +++ b/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") } }