diff --git a/CHANGELOG.md b/CHANGELOG.md index 261acf4..1da0efe 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,26 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## [Unreleased] +### Added +- `archive update` spaces its D2L requests out instead of firing them as + fast as the network returns them. Each gap is a randomized duration in + `[N, 2N)` rather than a fixed `N`, so consecutive requests don't land in + an evenly-spaced pattern. Tunable with `--request-interval`; the default + is 50ms and `--request-interval 0` disables pacing entirely. Because the + gap is measured from the previous request's start, a slow response + already satisfies the floor and costs no extra wall-clock. +- `archive update` now retries throttled and temporarily-failing requests + (HTTP 429 and 5xx) for up to 3 attempts with exponential backoff, jitter, + and `Retry-After` support in both delta-seconds and HTTP-date form. + A `Retry-After` hint is capped at 60s so a hostile or broken value cannot + stall a pass. Permanent failures (404, 400) and expired sessions are + still not retried. + +### Fixed +- `archive update` leaked a connection per TOC fetch and per course-list + page: the response body was decoded but never closed. Retries made those + leaks multiply, so they are closed now. + ## [0.3.0] - 2026-10-15 ### Added diff --git a/cmd/archive.go b/cmd/archive.go index 716f39a..dcd5b23 100644 --- a/cmd/archive.go +++ b/cmd/archive.go @@ -61,6 +61,7 @@ var ( archiveUpdateNoBodies bool archiveUpdateNoRefresh bool archiveUpdateWait bool + archiveUpdateInterval time.Duration archiveFindExt string archiveFindType string @@ -99,6 +100,7 @@ func init() { archiveUpdateCmd.Flags().BoolVar(&archiveUpdateNoBodies, "no-bodies", false, "Skip file-body downloads (metadata only)") archiveUpdateCmd.Flags().BoolVar(&archiveUpdateNoRefresh, "no-auto-refresh", false, "Do not re-run login if the saved session has expired") archiveUpdateCmd.Flags().BoolVar(&archiveUpdateWait, "wait", false, "Block instead of failing when another archive run holds the lock") + archiveUpdateCmd.Flags().DurationVar(&archiveUpdateInterval, "request-interval", archive.DefaultRequestInterval, "Floor on the gap between D2L requests; the real gap is randomized above it, so 0 disables pacing entirely") archiveFindCmd.Flags().StringVar(&archiveFindExt, "ext", "", "Filter by URL extension (e.g. pdf, docx)") archiveFindCmd.Flags().StringVar(&archiveFindType, "type", "", "Filter by topic type (File, Link)") @@ -252,6 +254,10 @@ var archiveUpdateCmd = &cobra.Command{ } func runArchiveUpdate(args []string) error { + if archiveUpdateInterval < 0 { + return fmt.Errorf("--request-interval must not be negative (got %s); pass 0 to disable pacing", archiveUpdateInterval) + } + abs, err := filepath.Abs(archiveDir) if err != nil { return err @@ -292,13 +298,14 @@ func runArchiveUpdate(args []string) error { scope := mergeScope(args, archiveCourses) start := time.Now() result, err := archive.Run(archive.Options{ - Root: abs, - Jar: ens.Jar, - Verbose: archiveVerbose, - Quiet: archiveJSON || archiveQuiet, - CourseIDs: scope, - Progress: progressAdapter(hud), - DownloadFiles: !archiveUpdateNoBodies, + Root: abs, + Jar: ens.Jar, + Verbose: archiveVerbose, + Quiet: archiveJSON || archiveQuiet, + CourseIDs: scope, + Progress: progressAdapter(hud), + DownloadFiles: !archiveUpdateNoBodies, + RequestInterval: &archiveUpdateInterval, }) if err != nil { return err @@ -360,12 +367,13 @@ func runArchiveUpdateDryRun(args []string, abs string) error { scope := mergeScope(args, archiveCourses) start := time.Now() result, err := archive.Diff(archive.Options{ - Root: abs, - Jar: ens.Jar, - Verbose: archiveVerbose, - Quiet: archiveJSON || archiveQuiet, - CourseIDs: scope, - DownloadFiles: !archiveUpdateNoBodies, + Root: abs, + Jar: ens.Jar, + Verbose: archiveVerbose, + Quiet: archiveJSON || archiveQuiet, + CourseIDs: scope, + DownloadFiles: !archiveUpdateNoBodies, + RequestInterval: &archiveUpdateInterval, }) if err != nil { return err diff --git a/internal/archive/archive.go b/internal/archive/archive.go index fca7a76..00aa1f2 100644 --- a/internal/archive/archive.go +++ b/internal/archive/archive.go @@ -301,6 +301,14 @@ type Options struct { CourseIDs []string // when non-empty, restrict to these course ids Now func() time.Time DownloadFiles bool // when true, also download each File topic's underlying bytes and hash-store them in blobs/.bin (default: false for backwards-compat) + // RequestInterval is the floor on the gap between consecutive outbound + // requests; the real gap is randomized above it. Nil means + // DefaultRequestInterval; an explicit 0 disables pacing altogether. + RequestInterval *time.Duration + + // pacer is the pass-wide rate limiter, created from RequestInterval on + // first use so one Run/Diff shares a single budget across all courses. + pacer *pacer } // baseURL returns the effective D2L base, defaulting to ua.D2LBase. @@ -522,6 +530,7 @@ func Run(opts Options) (Result, error) { if opts.Jar == nil { return Result{}, fmt.Errorf("archive: nil cookie jar") } + opts.pacer = opts.limiter() now := opts.Now if now == nil { now = func() time.Time { return time.Now().UTC() } @@ -563,7 +572,7 @@ func Run(opts Options) (Result, error) { if opts.Progress != nil { opts.Progress(ProgressEvent{Phase: "list"}) } - listed, err := listAllCoursesAt(opts.Jar, opts.baseURL()) + listed, err := listAllCoursesAt(opts, opts.baseURL()) if err != nil { return Result{}, fmt.Errorf("archive: list courses: %w", err) } @@ -652,6 +661,7 @@ func Diff(opts Options) (DiffResult, error) { if opts.Jar == nil { return DiffResult{}, fmt.Errorf("archive: nil cookie jar") } + opts.pacer = opts.limiter() result := DiffResult{Root: opts.Root} idx, err := LoadIndex(opts.Root) @@ -680,7 +690,7 @@ func Diff(opts Options) (DiffResult, error) { if opts.Progress != nil { opts.Progress(ProgressEvent{Phase: "list"}) } - listed, err := listAllCoursesAt(opts.Jar, opts.baseURL()) + listed, err := listAllCoursesAt(opts, opts.baseURL()) if err != nil { return result, fmt.Errorf("diff: list courses: %w", err) } @@ -733,7 +743,7 @@ func Diff(opts Options) (DiffResult, error) { // index, fetch the TOC, classify every topic, return the counts. No body // fetches, no index writes. func diffCourse(opts Options, c Course) (CourseDiff, error) { - toc, err := fetchTocFor(opts.baseURL(), c.OrgUnitId, opts.Jar) + toc, err := opts.fetchToc(opts.limiter(), c.OrgUnitId) if err != nil { return CourseDiff{}, err } @@ -806,7 +816,7 @@ func archiveCourse(opts Options, c Course, now func() time.Time) (courseStats, e func archiveCourseV2(opts Options, c Course, now func() time.Time) (courseStats, error) { var s courseStats transportFails := 0 - toc, err := fetchTocFor(opts.baseURL(), c.OrgUnitId, opts.Jar) + toc, err := opts.fetchToc(opts.limiter(), c.OrgUnitId) if err != nil { return s, err } @@ -873,7 +883,7 @@ func archiveCourseV2(opts Options, c Course, now func() time.Time) (courseStats, Reason: q.reason, Stale: s.stale, }) - body, ferr := fetchTopicFor(opts.baseURL(), opts.Jar, c.OrgUnitId, q.topicID) + body, ferr := opts.fetchTopic(opts.limiter(), c.OrgUnitId, q.topicID) if ferr != nil { if errors.Is(ferr, errSessionExpired) { return abortWithIndex(ferr) @@ -931,7 +941,7 @@ func archiveCourseV2(opts Options, c Course, now func() time.Time) (courseStats, if bodyURL == "" { continue } - bodyBytes, contentType, berr := downloadFileBody(opts.Jar, bodyURL, opts.baseURL()) + bodyBytes, contentType, berr := opts.fetchBody(opts.limiter(), bodyURL) if berr != nil { // The metadata for a topic whose reason was new/modified is // about to be recorded as current, but its body is not. Clear @@ -1061,7 +1071,8 @@ func downloadFileBody(jar *cookiejar.Jar, rawURL, base string) ([]byte, string, if len(msg) > 200 { msg = msg[:200] + "..." } - return nil, "", fmt.Errorf("download %d %s for %s: %s", res.StatusCode, res.Status, rawURL, msg) + return nil, "", newStatusError(res, + fmt.Sprintf("download %d %s for %s: %s", res.StatusCode, res.Status, rawURL, msg)) } // An expired session redirects the download to the D2L login page, // which answers 200 with HTML; that page must never be hashed as a @@ -1072,7 +1083,7 @@ func downloadFileBody(jar *cookiejar.Jar, rawURL, base string) ([]byte, string, } body, err := io.ReadAll(res.Body) if err != nil { - return nil, "", err + return nil, "", &transportError{err} } return body, res.Header.Get("Content-Type"), nil } @@ -1116,13 +1127,15 @@ func fetchTocFor(base, orgUnitID string, jar *cookiejar.Jar) (content.TocRespons Headers: map[string]string{"Accept": "application/json"}, }) if err != nil { - return content.TocResponse{}, err + return content.TocResponse{}, &transportError{err} } + defer func() { _ = res.Body.Close() }() if res.StatusCode != 200 { if res.StatusCode == 401 || res.StatusCode == 403 { return content.TocResponse{}, errSessionExpired } - return content.TocResponse{}, fmt.Errorf("D2L API returned %d %s for %s", res.StatusCode, res.Status, res.Request.URL) + return content.TocResponse{}, newStatusError(res, + fmt.Sprintf("D2L API returned %d %s for %s", res.StatusCode, res.Status, res.Request.URL)) } var toc content.TocResponse if err := json.NewDecoder(res.Body).Decode(&toc); err != nil { @@ -1145,14 +1158,15 @@ func fetchTopicFor(base string, jar *cookiejar.Jar, orgUnitID string, topicID in if res.StatusCode == 401 || res.StatusCode == 403 { return nil, errSessionExpired } - return nil, fmt.Errorf("D2L API returned %d %s for topic %d", res.StatusCode, res.Status, topicID) + return nil, newStatusError(res, + fmt.Sprintf("D2L API returned %d %s for topic %d", res.StatusCode, res.Status, topicID)) } if ct := res.Header.Get("Content-Type"); strings.Contains(ct, "text/html") || ua.IsLoginURL(res.Request.URL.String()) { return nil, errSessionExpired } body, err := httpclient.BodyBytes(res) if err != nil { - return nil, err + return nil, &transportError{err} } out, err := prettyJSON(body) if err != nil { @@ -1348,13 +1362,13 @@ type d2lCoursesResponse struct { Courses []d2lCourse `json:"Courses"` } -func listAllCoursesAt(jar *cookiejar.Jar, base string) ([]Course, error) { +func listAllCoursesAt(opts Options, base string) ([]Course, error) { var all []Course next := coursesAPIURLFor(base) pageNum := 0 for next != "" { pageNum++ - data, err := fetchCoursesPage(jar, next) + data, err := opts.fetchCoursesPage(opts.limiter(), next) if err != nil { return all, err } @@ -1377,13 +1391,15 @@ func listAllCoursesAt(jar *cookiejar.Jar, base string) ([]Course, error) { func fetchCoursesPage(jar *cookiejar.Jar, rawURL string) (d2lCoursesResponse, error) { res, err := httpclient.FollowRedirects(rawURL, jar, httpclient.FetchOptions{}) if err != nil { - return d2lCoursesResponse{}, err + return d2lCoursesResponse{}, &transportError{err} } + defer func() { _ = res.Body.Close() }() if res.StatusCode != 200 { if res.StatusCode == 401 || res.StatusCode == 403 { return d2lCoursesResponse{}, errSessionExpired } - return d2lCoursesResponse{}, fmt.Errorf("D2L my-courses API returned %d %s", res.StatusCode, res.Status) + return d2lCoursesResponse{}, newStatusError(res, + fmt.Sprintf("D2L my-courses API returned %d %s", res.StatusCode, res.Status)) } ct := res.Header.Get("Content-Type") if !strings.Contains(ct, "application/json") && !strings.Contains(ct, "text/json") { @@ -1391,7 +1407,7 @@ func fetchCoursesPage(jar *cookiejar.Jar, rawURL string) (d2lCoursesResponse, er } body, err := httpclient.BodyBytes(res) if err != nil { - return d2lCoursesResponse{}, err + return d2lCoursesResponse{}, &transportError{err} } var out d2lCoursesResponse if err := json.Unmarshal(body, &out); err != nil { diff --git a/internal/archive/pacing.go b/internal/archive/pacing.go new file mode 100644 index 0000000..8563dcc --- /dev/null +++ b/internal/archive/pacing.go @@ -0,0 +1,281 @@ +package archive + +import ( + "errors" + "math/rand" + "net/http" + "strconv" + "strings" + "sync" + "time" + + "github.com/aarav/schooltools/internal/content" + "github.com/aarav/schooltools/internal/retry" +) + +// DefaultRequestInterval is the floor on the wall-clock gap between two +// consecutive outbound D2L requests. +// +// Archive is strictly sequential, so the natural request rate is already +// bounded by round-trip latency — on a fast link the only thing separating +// two requests is how quickly the previous one came back. The floor keeps +// that from becoming the limit, and the jitter is what keeps the resulting +// traffic from landing in an evenly-spaced, machine-looking pattern. +// +// The default is deliberately liberal: real gaps land in +// [DefaultRequestInterval, 2*DefaultRequestInterval), and because the pacer +// measures from the previous request's start, a slow response already +// satisfies the floor on its own and costs no extra wall-clock at all. +// Users who want the old behaviour exactly can pass 0. +const DefaultRequestInterval = 50 * time.Millisecond + +// retryBaseDelayMs seeds the exponential backoff used for transient +// failures. retry.WithRetry owns the doubling and the jitter; this only +// sets the starting rung. +const retryBaseDelayMs = 250 + +// retryAttempts is deliberately small. Archive is a batch job the user can +// simply re-run, and the index is append-only, so a failed attempt is cheap +// to redo — there is no reason to sit in a long backoff loop. +const retryAttempts = 3 + +// pacer spaces out consecutive requests. It is safe for concurrent use, +// though the archive package does not fetch concurrently today. +type pacer struct { + mu sync.Mutex + floor time.Duration + last time.Time + + // Seams so tests can assert on the arithmetic without sleeping. + now func() time.Time + sleep func(time.Duration) + jitter func(time.Duration) time.Duration +} + +func newPacer(floor time.Duration) *pacer { + return &pacer{ + floor: floor, + now: time.Now, + sleep: time.Sleep, + jitter: randomGap, + } +} + +// Wait blocks until at least one randomised floor has elapsed since the +// previous call, then records this request as in flight. A nil pacer or a +// non-positive floor makes this a no-op. +func (p *pacer) Wait() { + if p == nil || p.floor <= 0 { + return + } + p.mu.Lock() + defer p.mu.Unlock() + target := p.floor + p.jitter(p.floor) + if !p.last.IsZero() { + if remaining := target - p.now().Sub(p.last); remaining > 0 { + p.sleep(remaining) + } + } + p.last = p.now() +} + +// randomGap draws a uniform duration in [0, floor). Drawing the extra gap +// fresh on every call is the whole point: a fixed floor on its own just +// trades an unthrottled stream for an evenly-spaced one, which is easier to +// fingerprint, not harder. +func randomGap(floor time.Duration) time.Duration { + if floor <= 0 { + return 0 + } + return time.Duration(rand.Int63n(int64(floor))) +} + +// statusError carries the HTTP status and any server-supplied retry hint so +// the retry predicate can tell a throttle from a permanent failure without +// matching on message text. msg is preserved verbatim so existing error +// output is unchanged. +type statusError struct { + status int + retryAfter time.Duration + msg string +} + +func (e *statusError) Error() string { return e.msg } + +func newStatusError(res *http.Response, msg string) *statusError { + return &statusError{ + status: res.StatusCode, + retryAfter: parseRetryAfter(res.Header.Get("Retry-After")), + msg: msg, + } +} + +// maxRetryAfter bounds how far a server-supplied Retry-After can stall a +// pass. Archive is strictly sequential, so honoring an unbounded hint — +// "Retry-After: 86400", or an HTTP-date far in the future — would hang the +// whole run with no progress output. Past the cap we retry on the normal +// backoff instead, which is the lesser evil. Clamping before converting to +// time.Duration also stops time.Duration(secs) * time.Second overflowing +// int64 and wrapping to a negative value the helper would then ignore. +const maxRetryAfter = 60 * time.Second + +// parseRetryAfter reads a Retry-After header in either the delta-seconds or +// HTTP-date form. An unparseable, absent, or over-cap value yields something +// that leaves the computed backoff in charge. +func parseRetryAfter(v string) time.Duration { + v = strings.TrimSpace(v) + if v == "" { + return 0 + } + if secs, err := strconv.Atoi(v); err == nil { + if secs >= 0 { + if secs > int(maxRetryAfter/time.Second) { + return maxRetryAfter + } + return time.Duration(secs) * time.Second + } + } else if errors.Is(err, strconv.ErrRange) && allDigits(v) { + // Digits that overflow an int64. The server asked for a wait far + // longer than we will ever honor, so clamp rather than fall through + // to the backoff and retry a throttle immediately. + return maxRetryAfter + } + if t, err := http.ParseTime(v); err == nil { + if d := time.Until(t); d > 0 { + return min(d, maxRetryAfter) + } + } + return 0 +} + +func allDigits(s string) bool { + if s == "" { + return false + } + for _, r := range s { + if r < '0' || r > '9' { + return false + } + } + return true +} + +// retryableError decides whether a failed archive request is worth another +// attempt. +// +// Only server-level throttling and server faults qualify. Transport errors +// are deliberately excluded: during an archive pass a transport failure +// almost always means the network is gone, and retrying it per-topic across +// a whole course is the exact behaviour the existing consecutive-transport- +// failure guard exists to stop. That guard aborts the pass promptly and is +// left in charge here. +// +// An expired session is excluded too — retrying it only burns attempts, and +// Run already treats it as an abort-the-pass signal. 404 is excluded for the +// same reason: the topic is gone, not flaky. +func retryableError(err error) bool { + if err == nil || errors.Is(err, errSessionExpired) { + return false + } + var se *statusError + if errors.As(err, &se) { + return se.status == http.StatusTooManyRequests || se.status >= 500 + } + return false +} + +// paced runs fn under the pass's pacer and retry policy. fn must release any +// response body it opens before returning, since a retry re-enters it. +// +// Retries are independent of the request interval: turning pacing off with +// --request-interval=0 still backs off on a transient failure, because +// retrying a broken connection faster is never the right call. +func (o Options) paced(p *pacer, fn func() error) error { + _, err := retry.WithRetry(func() (struct{}, error) { + p.Wait() + return struct{}{}, fn() + }, retry.Options{ + Attempts: retryAttempts, + BaseDelayMs: retryBaseDelayMs, + DelayFor: func(_ int, err error) time.Duration { + var se *statusError + if errors.As(err, &se) { + return se.retryAfter + } + return 0 + }, + }, retryableError) + return err +} + +// limiter returns the pass-wide pacer, creating it from Options on first use +// so Run and Diff can share one rate limit across every course rather than +// resetting it per course. Nil RequestInterval takes the default; an +// explicit 0 disables pacing. +func (o Options) limiter() *pacer { + if o.pacer != nil { + return o.pacer + } + floor := DefaultRequestInterval + if o.RequestInterval != nil { + switch { + case *o.RequestInterval > 0: + floor = *o.RequestInterval + case *o.RequestInterval == 0: + floor = 0 + } + // A negative interval is a typo. The CLI rejects one outright; + // for library callers, fall back to the default rather than + // silently disabling the protection pacing exists to provide. Zero + // stays the only way to turn pacing off. + } + return newPacer(floor) +} + +// ---- paced wrappers over the individual fetch helpers --------------------- +// +// The underlying fetch helpers stay free of pacing and retry logic; this is +// the single place a request acquires a gap and a retry policy. Keeping it +// here is what stops a change to one call site from accidentally pacing a +// request twice. + +func (o Options) fetchToc(p *pacer, orgUnitID string) (content.TocResponse, error) { + var out content.TocResponse + err := o.paced(p, func() error { + var ferr error + out, ferr = fetchTocFor(o.baseURL(), orgUnitID, o.Jar) + return ferr + }) + return out, err +} + +func (o Options) fetchTopic(p *pacer, orgUnitID string, topicID int) ([]byte, error) { + var out []byte + err := o.paced(p, func() error { + var ferr error + out, ferr = fetchTopicFor(o.baseURL(), o.Jar, orgUnitID, topicID) + return ferr + }) + return out, err +} + +func (o Options) fetchBody(p *pacer, rawURL string) ([]byte, string, error) { + var body []byte + var ct string + err := o.paced(p, func() error { + var ferr error + body, ct, ferr = downloadFileBody(o.Jar, rawURL, o.baseURL()) + return ferr + }) + return body, ct, err +} + +func (o Options) fetchCoursesPage(p *pacer, rawURL string) (d2lCoursesResponse, error) { + var out d2lCoursesResponse + err := o.paced(p, func() error { + var ferr error + out, ferr = fetchCoursesPage(o.Jar, rawURL) + return ferr + }) + return out, err +} diff --git a/internal/archive/pacing_test.go b/internal/archive/pacing_test.go new file mode 100644 index 0000000..bfafabf --- /dev/null +++ b/internal/archive/pacing_test.go @@ -0,0 +1,533 @@ +package archive + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/aarav/schooltools/internal/content" +) + +// fakeClock stands in for time.Now/time.Sleep so pacer arithmetic can be +// asserted exactly instead of within a tolerance wall-clock. +type fakeClock struct { + mu sync.Mutex + t time.Time + slept []time.Duration + sleptN int +} + +func newFakeClock() *fakeClock { + return &fakeClock{t: time.Unix(1_700_000_000, 0)} +} + +func (c *fakeClock) now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.t +} + +func (c *fakeClock) sleep(d time.Duration) { + c.mu.Lock() + defer c.mu.Unlock() + c.slept = append(c.slept, d) + c.sleptN++ + c.t = c.t.Add(d) +} + +// advance simulates time passing without a sleep, i.e. a slow response. +func (c *fakeClock) advance(d time.Duration) { + c.mu.Lock() + defer c.mu.Unlock() + c.t = c.t.Add(d) +} + +func (c *fakeClock) totalSlept() time.Duration { + c.mu.Lock() + defer c.mu.Unlock() + var sum time.Duration + for _, d := range c.slept { + sum += d + } + return sum +} + +// fixedJitter ignores the floor and returns a constant, so target arithmetic +// is exact: floor + 25ms. +func fixedJitter(time.Duration) time.Duration { return 25 * time.Millisecond } + +func TestPacerFirstRequestIsNotDelayed(t *testing.T) { + c := newFakeClock() + p := newPacer(50 * time.Millisecond) + p.now, p.sleep, p.jitter = c.now, c.sleep, fixedJitter + + p.Wait() + if c.sleptN != 0 { + t.Errorf("first Wait slept %d times, want 0 (nothing precedes it)", c.sleptN) + } +} + +func TestPacerEnforcesFloorPlusJitter(t *testing.T) { + c := newFakeClock() + p := newPacer(50 * time.Millisecond) + p.now, p.sleep, p.jitter = c.now, c.sleep, fixedJitter + + for i := 0; i < 5; i++ { + p.Wait() + } + // Every gap after the first must be floor + jitter = 75ms. + if want := 4 * 75 * time.Millisecond; c.totalSlept() != want { + t.Errorf("total slept = %v, want %v", c.totalSlept(), want) + } +} + +func TestPacerSkipsSleepWhenResponseAlreadySlow(t *testing.T) { + c := newFakeClock() + p := newPacer(50 * time.Millisecond) + p.now, p.sleep, p.jitter = c.now, c.sleep, fixedJitter + + p.Wait() + c.advance(500 * time.Millisecond) // a slow request already exceeded the floor + p.Wait() + if c.sleptN != 0 { + t.Errorf("pacer slept %d times despite a 500ms gap, want 0 (pacing must not add to a slow response)", c.sleptN) + } +} + +func TestPacerSleepsOnlyRemainingTime(t *testing.T) { + c := newFakeClock() + p := newPacer(50 * time.Millisecond) + p.now, p.sleep, p.jitter = c.now, c.sleep, fixedJitter + + p.Wait() + c.advance(60 * time.Millisecond) // target is 75ms, so 15ms is still owed + p.Wait() + if got := c.totalSlept(); got != 15*time.Millisecond { + t.Errorf("slept = %v, want 15ms (only the shortfall, not a fresh 75ms)", got) + } +} + +func TestPacerJitterVariesGaps(t *testing.T) { + c := newFakeClock() + p := newPacer(20 * time.Millisecond) + p.now, p.sleep, p.jitter = c.now, c.sleep, randomGap + + for i := 0; i < 60; i++ { + p.Wait() + } + + c.mu.Lock() + gaps := append([]time.Duration(nil), c.slept...) + c.mu.Unlock() + + distinct := map[time.Duration]struct{}{} + var belowFloor int + for _, g := range gaps { + distinct[g] = struct{}{} + if g < 20*time.Millisecond { + belowFloor++ + } + } + if belowFloor != 0 { + t.Errorf("%d/%d gaps fell under the 20ms floor", belowFloor, len(gaps)) + } + if len(distinct) < 5 { + t.Errorf("only %d distinct gap values across %d requests; jitter is not doing anything", len(distinct), len(gaps)) + } +} + +func TestPacerRandomGapStaysInRange(t *testing.T) { + floor := 30 * time.Millisecond + for i := 0; i < 500; i++ { + g := randomGap(floor) + if g < 0 || g >= floor { + t.Fatalf("randomGap(%v) = %v, want [0, %v)", floor, g, floor) + } + } + if randomGap(0) != 0 { + t.Errorf("randomGap(0) must be 0") + } +} + +func TestPacerZeroFloorNeverSleeps(t *testing.T) { + c := newFakeClock() + p := newPacer(0) + p.now, p.sleep, p.jitter = c.now, c.sleep, randomGap + + for i := 0; i < 10; i++ { + p.Wait() + } + if c.sleptN != 0 { + t.Errorf("zero-floor pacer slept %d times, want 0", c.sleptN) + } +} + +// TestPacerNilAndNegativeAreSafe asserts nothing: a panic in a Go test is +// already a failure, so calling both guards is the whole test. +func TestPacerNilAndNegativeAreSafe(_ *testing.T) { + var nilPacer *pacer + nilPacer.Wait() + newPacer(-time.Second).Wait() +} + +func TestOptionsLimiterIntervalResolution(t *testing.T) { + d := func(v time.Duration) *time.Duration { return &v } + + if got := (Options{}).limiter().floor; got != DefaultRequestInterval { + t.Errorf("nil RequestInterval -> floor %v, want %v", got, DefaultRequestInterval) + } + if got := (Options{RequestInterval: d(0)}).limiter().floor; got != 0 { + t.Errorf("explicit 0 -> floor %v, want 0 (pacing disabled)", got) + } + if got := (Options{RequestInterval: d(120 * time.Millisecond)}).limiter().floor; got != 120*time.Millisecond { + t.Errorf("120ms -> floor %v, want 120ms", got) + } + if got := (Options{RequestInterval: d(-time.Second)}).limiter().floor; got != DefaultRequestInterval { + t.Errorf("negative -> floor %v, want %v (a typo must not silently disable pacing)", got, DefaultRequestInterval) + } +} + +func TestParseRetryAfter(t *testing.T) { + if got := parseRetryAfter("5"); got != 5*time.Second { + t.Errorf("delta-seconds: got %v, want 5s", got) + } + if got := parseRetryAfter("0"); got != 0 { + t.Errorf("zero seconds: got %v, want 0", got) + } + if got := parseRetryAfter(""); got != 0 { + t.Errorf("empty: got %v, want 0", got) + } + if got := parseRetryAfter(" "); got != 0 { + t.Errorf("whitespace: got %v, want 0", got) + } + if got := parseRetryAfter("not-a-date"); got != 0 { + t.Errorf("garbage: got %v, want 0", got) + } + if got := parseRetryAfter("-3"); got != 0 { + t.Errorf("negative: got %v, want 0", got) + } + future := time.Now().Add(3 * time.Second).UTC().Format(http.TimeFormat) + if got := parseRetryAfter(future); got <= 0 || got > 3*time.Second { + t.Errorf("http-date: got %v, want roughly 3s", got) + } + past := time.Now().Add(-time.Hour).UTC().Format(http.TimeFormat) + if got := parseRetryAfter(past); got != 0 { + t.Errorf("past http-date: got %v, want 0", got) + } +} + +func TestParseRetryAfterIsCapped(t *testing.T) { + // An unbounded server hint must not be able to stall a sequential pass. + for _, v := range []string{"61", "86400", "10000000000", "99999999999999999999"} { + if got := parseRetryAfter(v); got != maxRetryAfter { + t.Errorf("parseRetryAfter(%q) = %v, want the %v cap", v, got, maxRetryAfter) + } + } + if got := parseRetryAfter("60"); got != maxRetryAfter { + t.Errorf("parseRetryAfter(\"60\") = %v, want %v (at the cap, not over)", got, maxRetryAfter) + } + if got := parseRetryAfter("59"); got != 59*time.Second { + t.Errorf("parseRetryAfter(\"59\") = %v, want 59s (under the cap, untouched)", got) + } + // A far-future HTTP-date must clamp rather than returning its full delta. + far := time.Now().Add(72 * time.Hour).UTC().Format(http.TimeFormat) + if got := parseRetryAfter(far); got != maxRetryAfter { + t.Errorf("parseRetryAfter(future date) = %v, want the %v cap", got, maxRetryAfter) + } + // A near-future date must survive intact. + near := time.Now().Add(5 * time.Second).UTC().Format(http.TimeFormat) + if got := parseRetryAfter(near); got <= 0 || got > 5*time.Second { + t.Errorf("parseRetryAfter(near date) = %v, want roughly 5s", got) + } +} + +func TestRetryableErrorClassification(t *testing.T) { + se := func(status int) error { return &statusError{status: status, msg: "x"} } + + cases := []struct { + name string + err error + want bool + }{ + {"nil", nil, false}, + {"429 throttled", se(429), true}, + {"500 server error", se(500), true}, + {"503 unavailable", se(503), true}, + {"404 gone", se(404), false}, + {"400 bad request", se(400), false}, + {"401 unauthenticated", se(401), false}, + {"session expired", errSessionExpired, false}, + {"transport failure", &transportError{err: context.DeadlineExceeded}, false}, + {"plain error", context.Canceled, false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := retryableError(tc.err); got != tc.want { + t.Errorf("retryableError(%v) = %v, want %v", tc.err, got, tc.want) + } + }) + } +} + +// throttleFixture serves a course whose single topic returns `failFirst` +// throttle responses before succeeding, and counts topic requests. +type throttleFixture struct { + calls int + failWith int + first int + header string +} + +func (tf *throttleFixture) wrap(inner http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if len(r.URL.Path) > 0 && containsTopicPath(r.URL.Path) { + tf.calls++ + if tf.calls <= tf.first { + if tf.header != "" { + w.Header().Set("Retry-After", tf.header) + } + w.WriteHeader(tf.failWith) + return + } + } + inner.ServeHTTP(w, r) + }) +} + +func containsTopicPath(p string) bool { + return strings.Contains(p, "/content/topics/") +} + +func singleTopicFixture(t *testing.T, tf *throttleFixture) (*httptest.Server, string) { + t.Helper() + f := newFakeD2L(t) + f.courses = []d2lCourse{{OrgUnitId: json.Number("5555"), Name: "Lab", Code: "LAB", IsActive: true}} + urlStr := "/content/enforced/5555-T1/notes.pdf" + f.tocs["5555"] = content.TocResponse{Modules: []content.TocModule{{ModuleId: 1, Topics: []content.TocTopic{ + {TopicId: 9500, Title: "Notes", LastModifiedDate: "2026-03-01T00:00:00Z", TypeIdentifier: "HTML"}, + }}}} + f.topics[9500] = map[string]any{"TopicId": 9500, "Title": "Notes", "TypeIdentifier": "HTML"} + _ = urlStr + + srv := httptest.NewServer(tf.wrap(f.handler())) + t.Cleanup(srv.Close) + return srv, "5555" +} + +func TestRunRetriesThrottledTopic(t *testing.T) { + tf := &throttleFixture{failWith: http.StatusTooManyRequests, first: 2} + srv, courseID := singleTopicFixture(t, tf) + + res, err := Run(Options{Root: t.TempDir(), Jar: newJar(t), BaseURL: srv.URL, CourseIDs: []string{courseID}}) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.TopicsFetched != 1 { + t.Errorf("TopicsFetched = %d, want 1 (a 429 must not lose the topic)", res.TopicsFetched) + } + if res.TopicsFailed != 0 { + t.Errorf("TopicsFailed = %d, want 0", res.TopicsFailed) + } + if tf.calls != 3 { + t.Errorf("topic requests = %d, want 3 (2 throttled + 1 success)", tf.calls) + } +} + +func TestRunRetriesServerErrorThenSucceeds(t *testing.T) { + tf := &throttleFixture{failWith: http.StatusBadGateway, first: 1} + srv, courseID := singleTopicFixture(t, tf) + + res, err := Run(Options{Root: t.TempDir(), Jar: newJar(t), BaseURL: srv.URL, CourseIDs: []string{courseID}}) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.TopicsFetched != 1 || tf.calls != 2 { + t.Errorf("fetched=%d calls=%d, want fetched=1 calls=2", res.TopicsFetched, tf.calls) + } +} + +func TestRunDoesNotRetryPermanentFailure(t *testing.T) { + tf := &throttleFixture{failWith: http.StatusNotFound, first: 99} + srv, courseID := singleTopicFixture(t, tf) + + res, err := Run(Options{Root: t.TempDir(), Jar: newJar(t), BaseURL: srv.URL, CourseIDs: []string{courseID}}) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.TopicsFetched != 0 { + t.Errorf("TopicsFetched = %d, want 0", res.TopicsFetched) + } + if res.TopicsFailed != 1 { + t.Errorf("TopicsFailed = %d, want 1", res.TopicsFailed) + } + if tf.calls != 1 { + t.Errorf("topic requests = %d, want exactly 1 (404 is permanent; retrying it is hammering)", tf.calls) + } +} + +func TestRunGivesUpAfterRetryBudget(t *testing.T) { + tf := &throttleFixture{failWith: http.StatusServiceUnavailable, first: 99} + srv, courseID := singleTopicFixture(t, tf) + + res, err := Run(Options{Root: t.TempDir(), Jar: newJar(t), BaseURL: srv.URL, CourseIDs: []string{courseID}}) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.TopicsFailed != 1 { + t.Errorf("TopicsFailed = %d, want 1", res.TopicsFailed) + } + if tf.calls != retryAttempts { + t.Errorf("topic requests = %d, want %d (the retry budget, then give up)", tf.calls, retryAttempts) + } +} + +func TestRunHonoursRetryAfter(t *testing.T) { + tf := &throttleFixture{failWith: http.StatusTooManyRequests, first: 1, header: "1"} + srv, courseID := singleTopicFixture(t, tf) + + start := time.Now() + res, err := Run(Options{Root: t.TempDir(), Jar: newJar(t), BaseURL: srv.URL, CourseIDs: []string{courseID}}) + elapsed := time.Since(start) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.TopicsFetched != 1 { + t.Errorf("TopicsFetched = %d, want 1", res.TopicsFetched) + } + if elapsed < time.Second { + t.Errorf("elapsed = %v, want >= 1s (Retry-After: 1 must outrank the 250ms backoff)", elapsed) + } +} + +// TestRunPacesRequestsAcrossCourses pins the property that the pacer is +// pass-wide: the gap between the last topic of one course and the TOC of the +// next must be paced too. A per-course pacer would leave that boundary +// unpaced, so the assertion is on the minimum observed gap across every +// request rather than on total elapsed time. +func TestRunPacesRequestsAcrossCourses(t *testing.T) { + f := newFakeD2L(t) + f.courses = []d2lCourse{ + {OrgUnitId: json.Number("1111"), Name: "A", Code: "A", IsActive: true}, + {OrgUnitId: json.Number("2222"), Name: "B", Code: "B", IsActive: true}, + } + + const topicsPerCourse = 6 + var received []time.Time + var mu sync.Mutex + mark := func() { + mu.Lock() + received = append(received, time.Now()) + mu.Unlock() + } + + for course, base := range map[string]int{"1111": 7000, "2222": 8000} { + var toc []content.TocTopic + for i := 0; i < topicsPerCourse; i++ { + id := base + i + toc = append(toc, content.TocTopic{ + TopicId: id, Title: "t", LastModifiedDate: "2026-03-01T00:00:00Z", TypeIdentifier: "HTML", + }) + f.topics[id] = map[string]any{"TopicId": id, "Title": "t", "TypeIdentifier": "HTML"} + } + f.tocs[course] = content.TocResponse{Modules: []content.TocModule{{ModuleId: 1, Topics: toc}}} + } + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.Contains(r.URL.Path, "/content/") || strings.Contains(r.URL.Path, "/mycourses") { + mark() + } + f.handler().ServeHTTP(w, r) + })) + defer srv.Close() + + interval := 20 * time.Millisecond + res, err := Run(Options{ + Root: t.TempDir(), + Jar: newJar(t), + BaseURL: srv.URL, + RequestInterval: &interval, + }) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.CoursesScanned != 2 { + t.Fatalf("CoursesScanned = %d, want 2", res.CoursesScanned) + } + if want := 2 * topicsPerCourse; res.TopicsFetched != want { + t.Fatalf("TopicsFetched = %d, want %d", res.TopicsFetched, want) + } + + mu.Lock() + defer mu.Unlock() + // 2 TOC requests + 12 topic requests + 1 course list. + if len(received) != 2*topicsPerCourse+3 { + t.Fatalf("recorded %d requests, want %d", len(received), 2*topicsPerCourse+3) + } + smallest := time.Duration(1<<62 - 1) + for i := 1; i < len(received); i++ { + if gap := received[i].Sub(received[i-1]); gap < smallest { + smallest = gap + } + } + // Slack absorbs measurement noise between two handler-side clock reads; + // a per-course pacer reset would show up here as a ~0ms boundary gap. + if floor := interval - 5*time.Millisecond; smallest < floor { + t.Errorf("smallest gap between consecutive requests = %v, want >= %v (pacing must survive the course boundary)", smallest, floor) + } +} + +func TestRunWithoutPacingIsFast(t *testing.T) { + f := newFakeD2L(t) + f.courses = []d2lCourse{{OrgUnitId: json.Number("1111"), Name: "C", Code: "C", IsActive: true}} + + const topics = 25 + var toc []content.TocTopic + for i := 0; i < topics; i++ { + id := 8000 + i + toc = append(toc, content.TocTopic{ + TopicId: id, Title: "t", LastModifiedDate: "2026-03-01T00:00:00Z", TypeIdentifier: "HTML", + }) + f.topics[id] = map[string]any{"TopicId": id, "Title": "t", "TypeIdentifier": "HTML"} + } + f.tocs["1111"] = content.TocResponse{Modules: []content.TocModule{{ModuleId: 1, Topics: toc}}} + + srv := httptest.NewServer(f.handler()) + defer srv.Close() + + zero := time.Duration(0) + // Inject a pacer that counts sleeps instead of sleeping, so this asserts + // "no waits were added" directly. A wall-clock ceiling would flake on a + // loaded machine without testing anything more precisely. + var mu sync.Mutex + var sleeps int + p := newPacer(zero) + p.sleep = func(time.Duration) { + mu.Lock() + sleeps++ + mu.Unlock() + } + + res, err := Run(Options{ + Root: t.TempDir(), + Jar: newJar(t), + BaseURL: srv.URL, + RequestInterval: &zero, + pacer: p, + }) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.TopicsFetched != topics { + t.Fatalf("TopicsFetched = %d, want %d", res.TopicsFetched, topics) + } + mu.Lock() + defer mu.Unlock() + if sleeps != 0 { + t.Errorf("pacer slept %d times with a zero interval, want 0", sleeps) + } +} diff --git a/internal/retry/retry.go b/internal/retry/retry.go index 1ce20fe..d7da0bd 100644 --- a/internal/retry/retry.go +++ b/internal/retry/retry.go @@ -19,6 +19,13 @@ var RetryableNetworkCodes = map[string]struct{}{ type Options struct { Attempts int BaseDelayMs int + // DelayFor lets an error raise the floor on the next wait, for cases + // where the server states its own retry hint (an HTTP Retry-After, a + // rate-limit Retry-After). It can only lengthen the computed backoff, + // never shorten it, so a caller cannot accidentally hammer sooner than + // the exponential schedule already allows. Returning 0 leaves the + // backoff untouched. Nil disables the hook. + DelayFor func(attempt int, err error) time.Duration } func DefaultOptions() Options { return Options{Attempts: 3, BaseDelayMs: 250} } @@ -69,12 +76,31 @@ func pow2(n int) float64 { func sleep(d time.Duration) { time.Sleep(d) } +// jitterDelay is the same random component backoff adds, in the same units. +// It exists so a server-supplied hint can raise the wait without flattening +// the randomization out of it. +func jitterDelay(base int) time.Duration { + if base <= 0 { + return 0 + } + return time.Duration(rand.Intn(base+1)) * time.Millisecond +} + // WithRetry calls fn up to opts.Attempts times, sleeping with exponential backoff // + jitter between attempts. shouldRetry decides whether a given error is worth // retrying; the default is any error containing known transient markers. +// +// The jitter in backoff matters as much as the backoff itself: without it +// every caller that fails at the same moment retries in lockstep and +// re-creates the thundering herd the backoff was meant to break up. func WithRetry[T any](fn func() (T, error), opts Options, shouldRetry func(error) bool) (T, error) { if opts.Attempts == 0 { - opts = DefaultOptions() + // Substitute field by field. Replacing the whole struct would drop a + // caller-supplied DelayFor, so a caller who passes only a hint would + // silently lose it. + def := DefaultOptions() + opts.Attempts = def.Attempts + opts.BaseDelayMs = def.BaseDelayMs } if shouldRetry == nil { shouldRetry = IsRetryable @@ -90,7 +116,17 @@ func WithRetry[T any](fn func() (T, error), opts Options, shouldRetry func(error if i == opts.Attempts-1 || !shouldRetry(err) { return zero, err } - sleep(backoff(i, opts.BaseDelayMs)) + d := backoff(i, opts.BaseDelayMs) + if opts.DelayFor != nil { + if hint := opts.DelayFor(i, err); hint > d { + // Treat the hint as a floor, not a replacement. Every client + // handed the same Retry-After value would otherwise wake at + // the same instant and recreate the burst the backoff was + // there to avoid, so the jitter goes back on top. + d = hint + jitterDelay(opts.BaseDelayMs) + } + } + sleep(d) } return zero, last } \ No newline at end of file diff --git a/internal/retry/retry_test.go b/internal/retry/retry_test.go index 977b202..14a90ce 100644 --- a/internal/retry/retry_test.go +++ b/internal/retry/retry_test.go @@ -69,4 +69,58 @@ func TestWithRetryExhaustsAttempts(t *testing.T) { assert.Equal(t, 3, calls) // Ensure it doesn't sleep too long for the test. _ = time.Millisecond +} + +func TestDelayForRaisesTheFloor(t *testing.T) { + calls := 0 + start := time.Now() + out, err := retry.WithRetry(func() (string, error) { + calls++ + if calls < 2 { + return "", errors.New("fetch failed") + } + return "ok", nil + }, retry.Options{ + Attempts: 2, + BaseDelayMs: 1, + DelayFor: func(int, error) time.Duration { return 300 * time.Millisecond }, + }, retry.IsRetryable) + assert.NoError(t, err) + assert.Equal(t, "ok", out) + assert.Equal(t, 2, calls) + assert.GreaterOrEqual(t, time.Since(start), 300*time.Millisecond) +} + +func TestDelayForCannotShortenBackoff(t *testing.T) { + calls := 0 + start := time.Now() + _, err := retry.WithRetry(func() (string, error) { + calls++ + return "", errors.New("fetch failed") + }, retry.Options{ + Attempts: 2, + BaseDelayMs: 300, + DelayFor: func(int, error) time.Duration { return time.Nanosecond }, + }, retry.IsRetryable) + assert.Error(t, err) + assert.Equal(t, 2, calls) + // Substituting the 1ns hint for the backoff would finish in ~0s. + assert.GreaterOrEqual(t, time.Since(start), 300*time.Millisecond) +} + +func TestDelayForSurvivesDefaultAttempts(t *testing.T) { + // Attempts is left at its zero value, so the defaults are substituted + // field by field. A whole-struct substitution would drop DelayFor here. + var gotHint bool + _, err := retry.WithRetry(func() (string, error) { + return "", errors.New("fetch failed") + }, retry.Options{ + BaseDelayMs: 1, + DelayFor: func(int, error) time.Duration { + gotHint = true + return time.Nanosecond + }, + }, retry.IsRetryable) + assert.Error(t, err) + assert.True(t, gotHint, "DelayFor was discarded when Attempts was defaulted") } \ No newline at end of file