diff --git a/CHANGELOG.md b/CHANGELOG.md index 18e44de..4a5013b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## Unreleased +- Isolate shared-head workflow failures without discarding unrelated completed items, stop acquisition and post-batch lookups at the first quota-reserve failure, and retain rolled-back item failures by requested family so metadata-only retries cannot clear failed children. +- Preserve completed issue and PR hydrations when another item fails, with atomic per-item writes, scoped failure resolution, and unchanged successful-sync coverage. Record issue and comment fetch failures in the existing ledger; resolve recovered parent failures even when child fetching still fails. `sync` and `fill-pr-details` return committed counts while exiting nonzero on incomplete acquisition, including quota stops. - Update SQLite's C-parser dependency to v4.29.6 and pin the automation guide's artifact-upload example to v7.0.1. - Add explicit cloud archive admission with immutable source, integrity, enrichment, and warning evidence while preserving strict publication defaults. Thanks @vincentkoc. - Compatibility: report portable publication time as `last_export_at`; `last_sync_at` now describes retained successful sync runs instead of old repository scan checkpoints. Thanks @obviyus. diff --git a/docs/commands.md b/docs/commands.md index 4293823..143aa9d 100644 --- a/docs/commands.md +++ b/docs/commands.md @@ -43,7 +43,7 @@ These work on every command. | Command | Purpose | Docs | | --- | --- | --- | | `gitcrawl sync owner/repo [--state --since --numbers --limit --include-comments --include-pr-details --with pr-details --progress-file --json]` | Sync issues and PRs from GitHub into local SQLite | [Sync](/sync/) | -| `gitcrawl sync-failures owner/repo [--include-resolved --limit N --json]` | List failed PR hydration attempts and optional resolved history | [Sync](/sync/#hydration-depth) | +| `gitcrawl sync-failures owner/repo [--include-resolved --limit N --json]` | List failed issue, comment, and PR hydration attempts and optional resolved history | [Sync](/sync/#hydration-depth) | | `gitcrawl coverage [owner/repo \| --repos owner/a,owner/b] [--min-missing-pr-details N --json]` | Report archive, PR-detail, and enrichment coverage/freshness | — | | `gitcrawl fill-pr-details owner/repo [--limit --order --batch-size --reserve-rate-limit --include-comments --json-progress --json]` | Hydrate locally missing pull request detail rows in bounded batches | — | | `gitcrawl capture owner/repo [--schema gitcrawl.capture.v1 --since RFC3339 --output path --json]` | Export a deterministic code-free conversation snapshot | [Conversation capture](/capture/) | @@ -59,6 +59,8 @@ request would cross that floor. The live probe observes other processes and tools that share the token. This is best-effort because an unrelated consumer can spend quota between the probe and request; the 1500 default provides concurrency headroom. Pass `--reserve-rate-limit N` to choose another floor. +An incomplete fill, including a quota stop, exits nonzero while returning the +counts already committed. See [partial failures](/sync/#partial-failures). For an end-to-end first-run sequence that combines `status --json`, `doctor --json`, `sync --numbers`, bounded `--sync-if-stale` search, `gitcrawl runs`, and Octopool live reads, see the [maintainer archive workflow](/maintainer-archive/). diff --git a/docs/sync.md b/docs/sync.md index b527af4..42c1e58 100644 --- a/docs/sync.md +++ b/docs/sync.md @@ -28,7 +28,7 @@ A sync writes: - `thread_revisions` — immutable revisions with content-addressed canonical evidence payloads when fully hydrated thread or review evidence changes - `thread_fingerprints` — one deterministic `thread-fingerprint-v2` row for each persisted revision - `documents` — canonical thread documents (when bodies change) -- `run_records` — sync run statistics +- `sync_runs` — sync run statistics and retry checkpoints Revision and fingerprint production fails closed on incomplete evidence. Issues require `--include-comments`; pull requests require both `--include-comments` and @@ -90,7 +90,8 @@ observation ordering. It does not fetch, clear, or mark files, commits, checks, workflow runs, or review-thread resolution as fresh. Comments remain independent: add `--include-comments` when needed. Selecting both hydration modes uses full `pr-details` hydration. Metadata-only hydration does not create full PR revisions -or fingerprints, and it resolves only earlier metadata-fetch failures. +or fingerprints. It resolves PR metadata-fetch failures, not failures of omitted +child collections. Full PR details also populate `pull_request_files`, `pull_request_commits`, `pull_request_checks`, and `github_workflow_runs` for local review and search. @@ -108,7 +109,40 @@ the incomplete review-thread response is not saved as complete evidence. Use `gitcrawl coverage [owner/repo] --json` to inspect archive completeness after a sync. It reports issue, PR, comment, and review counts alongside hydrated PR detail rows, missing PR details, known failed hydrations, and detail-table row counts per repository. The additive `enrichment` object exposes supported, eligible, covered, fresh, missing, stale, completeness, ratios, and latest timestamps for revisions, fingerprints, key summaries, clusters, and PR details. Use `--repos owner/a,owner/b` to compare selected repositories and `--min-missing-pr-details N` to focus backfill work on repositories with gaps. -`gitcrawl sync-failures owner/repo --json` lists unresolved PR hydration failures with their operation, error class and message, timestamps, and retry count. Add `--include-resolved` to inspect failures cleared by a later successful hydration. This operational ledger stays local when `portable prune` runs unless the publisher explicitly passes `--include-sync-failures`, which retains the ledger only after replacing every error message with a redaction marker. +`gitcrawl sync-failures owner/repo --json` lists unresolved issue, comment, and PR hydration failures with their operation, error class and message, timestamps, and retry count. Add `--include-resolved` to inspect failures cleared by a later successful hydration of that same family. This operational ledger stays local when `portable prune` runs unless the publisher explicitly passes `--include-sync-failures`, which retains the ledger only after replacing every error message with a redaction marker. + +## Partial failures + +Each completed issue or PR commits atomically with its requested children, +document, revision, fingerprint, and failure resolutions. A failed item does not +roll back completed siblings. Shared-head workflow observations are consolidated +before those writes, so sibling ordering cannot replace a newer snapshot. +A consolidation failure excludes only its shared-head group, records those +items as failed PR-detail hydrations, and preserves unrelated completed items. + +An item fetch failure records its actual operation: `issue`, `issue_comments`, +`pull_reviews`, `pull_review_comments`, `pull_review_threads`, +`pull_request_metadata`, or `pull_request_details`. A failed issue lookup does +not create a thread stub. Failed child fetches can retain the observed parent +metadata, but do not replace incomplete child collections or certify complete +evidence. Cancellation stops further work; transactions already committed remain. +A quota-reserve failure also stops new acquisition, including quota probes. +Completed payloads can still commit. Skipped requests are not recorded as +failures; shared-head groups that still need verification remain uncommitted. + +An item transaction failure records each requested operation after rollback, +with only an existing parent reference when available. It does not recreate the +rolled-back item. A retry resolves only the families it actually persists: +metadata-only retries leave comments and full-detail failures unresolved. + +An incomplete batch exits nonzero and never records a successful sync or advances +the closed-sweep watermark. Before partial writes, archives without a recorded +watermark retain their previous retry lower bound as a `checkpoint` in +`sync_runs`. A new archive uses the default 24-hour lower bound. This checkpoint +is not successful list coverage or freshness. Retry the failed numbers with the +same hydration flags after resolving the reported cause. +Older binaries can read these archives but do not honor the retry checkpoint +when writing. Do not downgrade the writer to resume a partially completed sync. `--include-code` is accepted for compatibility but is currently a no-op. @@ -177,6 +211,15 @@ stale-observation counts, the requested scope when present, and the database write destination. Use `gitcrawl runs owner/repo --kind sync --json` for recorded run IDs. +After partial persistence, `sync --json` still emits committed counts, but exits +nonzero and leaves progress marked `failed`. `fill-pr-details --json` likewise +reports committed `filled` and remaining selected items, including a partially +completed batch. A quota stop exits nonzero with +`stopped_reason: "rate-limit-reserve"`; other sync failures use `"sync-failed"`. +The partial batch and final result retain the stopping request's quota snapshot +without a subsequent credential or quota lookup. +Automation must check the exit status, not treat a JSON result as success. + ## Common workflows ### First-time setup for a repo diff --git a/internal/cli/app_test.go b/internal/cli/app_test.go index 148fd6c..a9644b6 100644 --- a/internal/cli/app_test.go +++ b/internal/cli/app_test.go @@ -6895,8 +6895,9 @@ func TestFillPRDetailsDefaultRateLimitFloorUsesLiveSharedQuota(t *testing.T) { run := New() var stdout bytes.Buffer run.Stdout = &stdout - if err := run.Run(ctx, []string{"--config", configPath, "fill-pr-details", "openclaw/gitcrawl", "--limit", "1", "--json"}); err != nil { - t.Fatalf("fill-pr-details: %v", err) + var reserveErr *gh.RateLimitReserveError + if err := run.Run(ctx, []string{"--config", configPath, "fill-pr-details", "openclaw/gitcrawl", "--limit", "1", "--json"}); !errors.As(err, &reserveErr) { + t.Fatalf("incomplete fill must return quota failure: %v", err) } var result struct { Selected int `json:"selected"` @@ -7023,8 +7024,9 @@ func TestFillPRDetailsReserveRateLimitStopsBeforeCrossingDuringBatch(t *testing. run := New() var stdout bytes.Buffer run.Stdout = &stdout - if err := run.Run(ctx, []string{"--config", configPath, "fill-pr-details", "openclaw/gitcrawl", "--limit", "1", "--reserve-rate-limit", "10", "--json"}); err != nil { - t.Fatalf("fill-pr-details: %v", err) + var reserveErr *gh.RateLimitReserveError + if err := run.Run(ctx, []string{"--config", configPath, "fill-pr-details", "openclaw/gitcrawl", "--limit", "1", "--reserve-rate-limit", "10", "--json"}); !errors.As(err, &reserveErr) { + t.Fatalf("incomplete fill must return quota failure: %v", err) } var result struct { Selected int `json:"selected"` diff --git a/internal/cli/sync.go b/internal/cli/sync.go index b8b8112..fef5348 100644 --- a/internal/cli/sync.go +++ b/internal/cli/sync.go @@ -76,20 +76,19 @@ func (a *App) runSync(ctx context.Context, args []string) error { IncludePRDetails: *includePRDetails || with["pr-details"], Progress: progress.report, }) + terminalState := syncProgressSucceeded if err != nil { - if progressErr := progress.finish(syncProgressFailed); progressErr != nil { - return progressErr - } - return err + terminalState = syncProgressFailed } - if err := progress.finish(syncProgressSucceeded); err != nil { + err = errors.Join(err, progress.finish(terminalState)) + if stats.Repository == "" { return err } result := struct { syncer.Stats dbTargetInfo }{Stats: stats, dbTargetInfo: target} - return a.writeOutput("sync", result, true) + return errors.Join(err, a.writeOutput("sync", result, true)) } type syncOptions struct { @@ -208,6 +207,7 @@ func (a *App) runFillPRDetails(ctx context.Context, args []string) error { ReserveRateLimit: reserve, dbTargetInfo: rt.dbTarget(), } + var syncErr error for i := 0; i < len(numbers); i += batchSize { end := i + batchSize if end > len(numbers) { @@ -229,17 +229,16 @@ func (a *App) runFillPRDetails(ctx context.Context, args []string) error { Quiet: *jsonProgress, RateLimitReserve: reserve, }) - if err != nil { - var reserveErr *gh.RateLimitReserveError - if errors.As(err, &reserveErr) { - rate := fillRateLimitResultFromSnapshot(reserveErr.RateLimit, reserve) - result.StoppedReason = "rate-limit-reserve" - result.RateLimit = &rate - break - } - return err + var reserveErr *gh.RateLimitReserveError + var rate fillRateLimitResult + var hasRate bool + if errors.As(err, &reserveErr) { + // The stopping request owns this snapshot; do not resolve credentials + // or inspect a different resource after the quota guard has fired. + rate, hasRate = fillRateLimitResultFromSnapshot(reserveErr.RateLimit, reserve), true + } else if err == nil { + rate, hasRate = a.currentFillRateLimit(ctx, reserve) } - rate, hasRate := a.currentFillRateLimit(ctx, reserve) batch := fillPRDetailsBatch{ Index: len(result.Batches) + 1, Numbers: batchNumbers, @@ -250,8 +249,18 @@ func (a *App) runFillPRDetails(ctx context.Context, args []string) error { batch.RateLimit = &rate result.RateLimit = &rate } - result.Batches = append(result.Batches, batch) + if err == nil || stats.PRDetailsSynced > 0 { + result.Batches = append(result.Batches, batch) + } result.Filled += stats.PRDetailsSynced + if err != nil { + syncErr = err + result.StoppedReason = "sync-failed" + if reserveErr != nil { + result.StoppedReason = "rate-limit-reserve" + } + break + } if *jsonProgress { a.writeFillPRDetailsProgress(fillPRDetailsProgressEvent{ Event: "batch_done", @@ -268,7 +277,7 @@ func (a *App) runFillPRDetails(ctx context.Context, args []string) error { result.Remaining = 0 } result.FinishedAt = time.Now().UTC().Format(time.RFC3339Nano) - return a.writeOutput("fill-pr-details", result, true) + return errors.Join(syncErr, a.writeOutput("fill-pr-details", result, true)) } type fillPRDetailsProgressEvent struct { @@ -426,8 +435,5 @@ func (a *App) syncRepository(ctx context.Context, owner, repo string, options sy Logger: logger, Progress: options.Progress, }) - if err != nil { - return syncer.Stats{}, target, err - } - return stats, target, nil + return stats, target, err } diff --git a/internal/cli/sync_partial_test.go b/internal/cli/sync_partial_test.go new file mode 100644 index 0000000..3838214 --- /dev/null +++ b/internal/cli/sync_partial_test.go @@ -0,0 +1,248 @@ +package cli + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strconv" + "strings" + "sync/atomic" + "testing" + "time" + + gh "github.com/openclaw/gitcrawl/internal/github" + "github.com/openclaw/gitcrawl/internal/store" + "github.com/openclaw/gitcrawl/internal/syncer" +) + +func TestSyncCommandsReturnCommittedCountsAndFailure(t *testing.T) { + for _, command := range []string{"sync", "fill-pr-details"} { + t.Run(command, func(t *testing.T) { + ctx := context.Background() + dir := t.TempDir() + configPath, dbPath := filepath.Join(dir, "config.toml"), filepath.Join(dir, "archive.db") + if err := New().Run(ctx, []string{"--config", configPath, "init", "--db", dbPath}); err != nil { + t.Fatal(err) + } + configureTestGitHubCache(t, configPath, filepath.Join(dir, "cache"), "test-token") + if command == "fill-pr-details" { + st, err := store.Open(ctx, dbPath) + if err != nil { + t.Fatal(err) + } + repoID, err := st.UpsertRepository(ctx, store.Repository{ + Owner: "fixture", Name: "repo", FullName: "fixture/repo", RawJSON: "{}", UpdatedAt: "2026-04-01T00:00:00Z", + }) + if err != nil { + t.Fatal(err) + } + for _, number := range []int{101, 102, 103} { + if _, err := st.UpsertThread(ctx, store.Thread{ + RepoID: repoID, GitHubID: strconv.Itoa(number + 10000), Number: number, Kind: "pull_request", State: "open", + Title: "fixture", LabelsJSON: "[]", AssigneesJSON: "[]", RawJSON: "{}", + ContentHash: fmt.Sprint(number), UpdatedAt: "2026-04-01T00:00:00Z", + }); err != nil { + t.Fatal(err) + } + } + if err := st.Close(); err != nil { + t.Fatal(err) + } + } + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("X-RateLimit-Limit", "5000") + w.Header().Set("X-RateLimit-Remaining", "4990") + w.Header().Set("X-RateLimit-Reset", strconv.FormatInt(time.Now().Add(time.Hour).Unix(), 10)) + path := r.URL.Path + switch { + case path == "/rate_limit": + _ = json.NewEncoder(w).Encode(map[string]any{"resources": map[string]any{ + "core": map[string]any{"limit": 5000, "remaining": 4990, "reset": time.Now().Add(time.Hour).Unix()}, + "graphql": map[string]any{"limit": 5000, "remaining": 4990, "reset": time.Now().Add(time.Hour).Unix()}, + }}) + case path == "/repos/fixture/repo": + _ = json.NewEncoder(w).Encode(map[string]any{"id": 123, "full_name": "fixture/repo"}) + case path == "/repos/fixture/repo/issues/102": + http.Error(w, "unavailable", http.StatusNotFound) + case path == "/repos/fixture/repo/issues/101" || path == "/repos/fixture/repo/issues/103": + number, _ := strconv.Atoi(path[strings.LastIndex(path, "/")+1:]) + kind := "issue" + if command == "fill-pr-details" { + kind = "pull_request" + } + _ = json.NewEncoder(w).Encode(githubIssueJSON(number, kind, "fixture")) + case strings.HasSuffix(path, "/comments"), strings.HasSuffix(path, "/files"), strings.HasSuffix(path, "/commits"): + _ = json.NewEncoder(w).Encode([]any{}) + case path == "/repos/fixture/repo/pulls/101" || path == "/repos/fixture/repo/pulls/103": + _ = json.NewEncoder(w).Encode(map[string]any{"head": map[string]any{"sha": ""}}) + case path == "/graphql": + _ = json.NewEncoder(w).Encode(map[string]any{"data": map[string]any{ + "repository": map[string]any{"pullRequest": map[string]any{"reviewThreads": map[string]any{ + "nodes": []any{}, "pageInfo": map[string]any{"hasNextPage": false, "endCursor": ""}, + }}}, + }}) + default: + t.Errorf("unexpected request: %s", path) + http.Error(w, "unexpected request", http.StatusBadRequest) + } + })) + defer server.Close() + t.Setenv("GITHUB_TOKEN", "test-token") + t.Setenv("GITCRAWL_GITHUB_BASE_URL", server.URL) + progressPath := filepath.Join(dir, "progress.json") + args := []string{"--config", configPath, command, "fixture/repo", "--json"} + if command == "sync" { + args = append(args, "--numbers", "101,102,103", "--include-comments", "--progress-file", progressPath) + } else { + args = append(args, "--batch-size", "3", "--reserve-rate-limit", "10") + } + run := New() + var stdout, stderr bytes.Buffer + run.Stdout, run.Stderr = &stdout, &stderr + err := run.Run(ctx, args) + if err == nil || ExitCode(err) == 0 { + t.Fatalf("incomplete acquisition reported success: %v", err) + } + if command == "sync" { + var stats syncer.Stats + if err := json.Unmarshal(stdout.Bytes(), &stats); err != nil { + t.Fatalf("result=%s err=%v", stdout.String(), err) + } + if stats.ThreadsSynced != 2 || stats.ClosedSweepThrough != "" { + t.Fatalf("committed stats=%+v", stats) + } + data, err := os.ReadFile(progressPath) + if err != nil { + t.Fatal(err) + } + var progress syncProgressSnapshot + if err := json.Unmarshal(data, &progress); err != nil || progress.State != syncProgressFailed { + t.Fatalf("terminal progress=%+v err=%v", progress, err) + } + } else { + var result fillPRDetailsResult + if err := json.Unmarshal(stdout.Bytes(), &result); err != nil { + t.Fatalf("result=%s err=%v", stdout.String(), err) + } + if result.Filled != 2 || result.Remaining != 1 || result.Selected != 3 || result.StoppedReason != "sync-failed" || + len(result.Batches) != 1 || result.Batches[0].PRDetailsSynced != 2 { + t.Fatalf("committed fill result=%+v err=%v stderr=%s", result, err, stderr.String()) + } + } + }) + } +} + +func TestFillQuotaStopReturnsCommittedCountsWithoutPostBatchLookup(t *testing.T) { + ctx := context.Background() + dir := t.TempDir() + configPath, dbPath := filepath.Join(dir, "config.toml"), filepath.Join(dir, "archive.db") + if err := New().Run(ctx, []string{"--config", configPath, "init", "--db", dbPath}); err != nil { + t.Fatal(err) + } + configureTestGitHubCache(t, configPath, filepath.Join(dir, "cache"), "") + st, err := store.Open(ctx, dbPath) + if err != nil { + t.Fatal(err) + } + repoID, err := st.UpsertRepository(ctx, store.Repository{ + Owner: "fixture", Name: "repo", FullName: "fixture/repo", RawJSON: "{}", UpdatedAt: "2026-04-01T00:00:00Z", + }) + if err != nil { + t.Fatal(err) + } + for _, number := range []int{101, 102} { + if _, err := st.UpsertThread(ctx, store.Thread{ + RepoID: repoID, GitHubID: strconv.Itoa(number), Number: number, Kind: "pull_request", State: "open", + Title: "fixture", LabelsJSON: "[]", AssigneesJSON: "[]", RawJSON: "{}", + ContentHash: fmt.Sprint(number), UpdatedAt: "2026-04-01T00:00:00Z", + }); err != nil { + t.Fatal(err) + } + } + if err := st.Close(); err != nil { + t.Fatal(err) + } + var completed, stopped atomic.Bool + var authCalls, postStopRequests atomic.Int32 + resetAt := time.Now().Add(time.Hour).Truncate(time.Second) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if stopped.Load() { + postStopRequests.Add(1) + http.Error(w, "request after quota stop", http.StatusBadRequest) + return + } + path := r.URL.Path + switch { + case path == "/rate_limit": + remaining := 100 + if completed.Load() { + stopped.Store(true) + remaining = 10 + } + _ = json.NewEncoder(w).Encode(map[string]any{"resources": map[string]any{ + "core": map[string]any{"limit": 5000, "remaining": remaining, "reset": resetAt.Unix()}, + "graphql": map[string]any{"limit": 5000, "remaining": remaining, "reset": resetAt.Unix()}, + }}) + case path == "/repos/fixture/repo": + _ = json.NewEncoder(w).Encode(map[string]any{"id": 123, "full_name": "fixture/repo"}) + case path == "/repos/fixture/repo/issues/101" || path == "/repos/fixture/repo/issues/102": + number, _ := strconv.Atoi(path[strings.LastIndex(path, "/")+1:]) + _ = json.NewEncoder(w).Encode(githubIssueJSON(number, "pull_request", "fixture")) + case path == "/graphql": + _ = json.NewEncoder(w).Encode(map[string]any{"data": map[string]any{ + "repository": map[string]any{"pullRequest": map[string]any{"reviewThreads": map[string]any{ + "nodes": []any{}, "pageInfo": map[string]any{"hasNextPage": false, "endCursor": ""}, + }}}, + }}) + case strings.HasSuffix(path, "/files"): + _ = json.NewEncoder(w).Encode([]any{}) + case strings.HasSuffix(path, "/commits"): + completed.Store(true) + _ = json.NewEncoder(w).Encode([]any{}) + case path == "/repos/fixture/repo/pulls/101" || path == "/repos/fixture/repo/pulls/102": + _ = json.NewEncoder(w).Encode(map[string]any{"head": map[string]any{"sha": ""}}) + default: + t.Errorf("unexpected request: %s", path) + http.Error(w, "unexpected request", http.StatusBadRequest) + } + })) + defer server.Close() + t.Setenv("GITHUB_TOKEN", "") + t.Setenv("GITCRAWL_GITHUB_BASE_URL", server.URL) + run := New() + run.githubAuthTokenLookup = func(context.Context) (string, error) { + authCalls.Add(1) + return "fixture-token", nil + } + var stdout, stderr bytes.Buffer + run.Stdout, run.Stderr = &stdout, &stderr + err = run.Run(ctx, []string{"--config", configPath, "fill-pr-details", "fixture/repo", "--batch-size", "2", "--reserve-rate-limit", "10", "--json"}) + var reserveErr *gh.RateLimitReserveError + if !errors.As(err, &reserveErr) || ExitCode(err) == 0 { + t.Fatalf("quota error lost: %v", err) + } + var result fillPRDetailsResult + if err := json.Unmarshal(stdout.Bytes(), &result); err != nil { + t.Fatalf("result=%s err=%v", stdout.String(), err) + } + if result.Filled != 1 || result.Remaining != 1 || result.StoppedReason != "rate-limit-reserve" || + len(result.Batches) != 1 || result.Batches[0].PRDetailsSynced != 1 { + t.Fatalf("partial quota result=%+v", result) + } + wantRate := fillRateLimitResultFromSnapshot(reserveErr.RateLimit, 10) + if result.RateLimit == nil || *result.RateLimit != wantRate || + result.Batches[0].RateLimit == nil || *result.Batches[0].RateLimit != wantRate { + t.Fatalf("blocking quota snapshot lost: result=%+v batch=%+v want=%+v", + result.RateLimit, result.Batches[0].RateLimit, wantRate) + } + if !stopped.Load() || postStopRequests.Load() != 0 || authCalls.Load() != 1 { + t.Fatalf("post-stop work: stopped=%t requests=%d auth=%d", stopped.Load(), postStopRequests.Load(), authCalls.Load()) + } +} diff --git a/internal/store/runs.go b/internal/store/runs.go index edd3b96..9c63bdd 100644 --- a/internal/store/runs.go +++ b/internal/store/runs.go @@ -3,6 +3,7 @@ package store import ( "context" "database/sql" + "encoding/json" "fmt" "strings" "time" @@ -128,21 +129,57 @@ func (s *Store) LastSuccessfulSyncAt(ctx context.Context, repoID int64) (time.Ti return parsed, nil } -// ClosedSweepWatermark ignores runs that did not establish complete default -// coverage. Old archives bootstrap from their oldest still-open observation. +const recordedClosedSweepWatermarkSQL = ` + select json_extract(stats_json, '$.closed_sweep_through') + from sync_runs + where repo_id = ? and status in ('success', 'completed', 'checkpoint') + and scope in ('open', 'closed', 'all') + and json_valid(stats_json) + and json_type(case when json_valid(stats_json) then stats_json else '{}' end, + '$.closed_sweep_through') = 'text' + order by julianday(json_extract(stats_json, '$.closed_sweep_through')) desc, id desc + limit 1 +` + +// PreserveClosedSweepWatermark freezes the legacy lower bound before partial +// commits can close or refresh its oldest open thread. A checkpoint is not a +// successful sync and cannot certify list freshness or archive completeness. +// The caller owns the transaction that precedes thread mutations. +func (s *Store) PreserveClosedSweepWatermark(ctx context.Context, repoID int64, startedAt time.Time) error { + var recorded string + err := s.q().QueryRowContext(ctx, recordedClosedSweepWatermarkSQL, repoID).Scan(&recorded) + if err == nil { + return nil + } + if err != sql.ErrNoRows { + return err + } + watermark, err := s.ClosedSweepWatermark(ctx, repoID) + if err != nil { + return err + } + if watermark.IsZero() { + watermark = startedAt.Add(-24 * time.Hour) + } + stats, err := json.Marshal(map[string]string{"closed_sweep_through": watermark.Format(time.RFC3339Nano)}) + if err != nil { + return err + } + _, err = s.RecordRun(ctx, RunRecord{ + RepoID: repoID, Kind: "sync", Scope: "open", Status: "checkpoint", + StartedAt: startedAt.Format(time.RFC3339Nano), + FinishedAt: startedAt.Format(time.RFC3339Nano), StatsJSON: string(stats), + }) + return err +} + +// ClosedSweepWatermark ignores incomplete runs. A legacy checkpoint retains +// only the old retry lower bound; complete default syncs advance it. func (s *Store) ClosedSweepWatermark(ctx context.Context, repoID int64) (time.Time, error) { var raw sql.NullString err := s.q().QueryRowContext(ctx, ` select coalesce( - (select json_extract(stats_json, '$.closed_sweep_through') - from sync_runs - where repo_id = ? and status in ('success', 'completed') - and scope in ('open', 'closed', 'all') - and json_valid(stats_json) - and json_type(case when json_valid(stats_json) then stats_json else '{}' end, - '$.closed_sweep_through') = 'text' - order by julianday(json_extract(stats_json, '$.closed_sweep_through')) desc, id desc - limit 1), + (`+recordedClosedSweepWatermarkSQL+`), (select min(coalesce(nullif(last_pulled_at, ''), nullif(first_pulled_at, ''), nullif(updated_at_gh, ''), updated_at)) from threads where repo_id = ? and state = 'open') diff --git a/internal/syncer/closed_sweep_test.go b/internal/syncer/closed_sweep_test.go index 05afe24..f4d80d9 100644 --- a/internal/syncer/closed_sweep_test.go +++ b/internal/syncer/closed_sweep_test.go @@ -216,3 +216,104 @@ func TestClosedSweepPreservesExplicitScopeContracts(t *testing.T) { }) } } + +func TestLegacyCheckpointSurvivesCommittedClosureAndFailedRunRecord(t *testing.T) { + ctx := context.Background() + path := filepath.Join(t.TempDir(), "archive.db") + st, err := store.Open(ctx, path) + if err != nil { + t.Fatal(err) + } + defer func() { _ = st.Close() }() + client := &defaultSweepGitHub{} + s := New(client, st) + started := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + now := started + s.now = func() time.Time { return now } + opts := Options{Owner: "fixture", Repo: "repo"} + if _, err := s.Sync(ctx, opts); err != nil { + t.Fatal(err) + } + repo, err := st.RepositoryByFullName(ctx, "fixture/repo") + if err != nil { + t.Fatal(err) + } + if _, err := st.DB().ExecContext(ctx, `update sync_runs set stats_json = '{}'`); err != nil { + t.Fatal(err) + } + if _, err := st.DB().ExecContext(ctx, `create trigger fail_success before insert on sync_runs + when new.status = 'success' + begin select raise(abort, 'final run rejected'); end`); err != nil { + t.Fatal(err) + } + now = now.Add(60 * 24 * time.Hour) + client.closed, client.updated = true, now.Format(time.RFC3339Nano) + stats, err := s.Sync(ctx, opts) + if err == nil || stats.ThreadsClosed != 1 || stats.ThreadsSynced != 1 || stats.ClosedSweepThrough != "" { + t.Fatalf("final record failure lost truthful commits: %+v err=%v", stats, err) + } + threads, err := st.ListThreads(ctx, repo.ID, true) + if err != nil || len(threads) != 1 || threads[0].State != "closed" { + t.Fatalf("committed closure lost=%+v err=%v", threads, err) + } + if err := st.Close(); err != nil { + t.Fatal(err) + } + st, err = store.Open(ctx, path) + if err != nil { + t.Fatal(err) + } + watermark, err := st.ClosedSweepWatermark(ctx, repo.ID) + if err != nil || !watermark.Equal(started) { + t.Fatalf("legacy checkpoint lost after reopen=%v err=%v", watermark, err) + } + last, err := st.LastSuccessfulSyncAt(ctx, repo.ID) + if err != nil || !last.Equal(started) { + t.Fatalf("failed final record advanced freshness=%v err=%v", last, err) + } + if _, err := st.DB().ExecContext(ctx, `drop trigger fail_success`); err != nil { + t.Fatal(err) + } + s = New(client, st) + s.now = func() time.Time { return now } + if _, err := s.Sync(ctx, opts); err != nil { + t.Fatal(err) + } + if got := client.requests[len(client.requests)-1].Since; got != started.Add(-time.Minute).Format(time.RFC3339Nano) { + t.Fatalf("retry lost original window=%q", got) + } + watermark, err = st.ClosedSweepWatermark(ctx, repo.ID) + if err != nil || !watermark.Equal(now) { + t.Fatalf("complete retry did not advance=%v err=%v", watermark, err) + } +} + +func TestNewArchivePartialCommitPreservesInitialSweepLowerBound(t *testing.T) { + ctx := context.Background() + st, err := store.Open(ctx, filepath.Join(t.TempDir(), "archive.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + client := &partialGitHub{failNumber: 8, operation: "issue"} + s := New(client, st) + started := time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC) + s.now = func() time.Time { return started } + stats, err := s.Sync(ctx, Options{Owner: "fixture", Repo: "repo", Numbers: []int{7, 8}}) + if err == nil || stats.ThreadsSynced != 1 { + t.Fatalf("partial sync=%+v err=%v", stats, err) + } + repo, err := st.RepositoryByFullName(ctx, "fixture/repo") + if err != nil { + t.Fatal(err) + } + watermark, err := st.ClosedSweepWatermark(ctx, repo.ID) + if err != nil || !watermark.Equal(started.Add(-24*time.Hour)) { + t.Fatalf("partial observation advanced initial lower bound=%v err=%v", watermark, err) + } + assertNoSuccessfulSync(t, st, repo.ID) + runs, err := st.ListRuns(ctx, repo.ID, "sync", 10) + if err != nil || len(runs) != 1 || runs[0].Status != "checkpoint" { + t.Fatalf("checkpoint=%+v err=%v", runs, err) + } +} diff --git a/internal/syncer/failure_isolation_test.go b/internal/syncer/failure_isolation_test.go new file mode 100644 index 0000000..759516e --- /dev/null +++ b/internal/syncer/failure_isolation_test.go @@ -0,0 +1,574 @@ +package syncer + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "path/filepath" + "slices" + "strconv" + "strings" + "testing" + "time" + + gh "github.com/openclaw/gitcrawl/internal/github" + "github.com/openclaw/gitcrawl/internal/store" +) + +type isolatedWorkflowGitHub struct { + sameSyncSharedHeadGitHub + failure string +} + +func (f *isolatedWorkflowGitHub) GetIssue(ctx context.Context, owner, repo string, number int, reporter gh.Reporter) (map[string]any, error) { + if number == 7 { + return f.fakeGitHub.GetIssue(ctx, owner, repo, number, reporter) + } + return f.sameSyncSharedHeadGitHub.GetIssue(ctx, owner, repo, number, reporter) +} + +func (f *isolatedWorkflowGitHub) GetPull(ctx context.Context, owner, repo string, number int, reporter gh.Reporter) (map[string]any, error) { + row, err := f.sameSyncSharedHeadGitHub.GetPull(ctx, owner, repo, number, reporter) + if number == 10 { + row["head"].(map[string]any)["sha"] = "other-head" + } + return row, err +} + +func (f *isolatedWorkflowGitHub) ListWorkflowRuns(ctx context.Context, owner, repo string, options gh.ListWorkflowRunsOptions, reporter gh.Reporter) ([]map[string]any, error) { + if options.HeadSHA == "other-head" { + return []map[string]any{{"id": 1000, "head_sha": "other-head", "created_at": "2026-07-12T00:00:00Z", "updated_at": "2026-07-12T00:01:00Z"}}, nil + } + rows, err := f.sameSyncSharedHeadGitHub.ListWorkflowRuns(ctx, owner, repo, options, reporter) + if f.failure == "conflict" && f.runCalls == 2 { + for _, row := range rows { + if jsonID(row["id"]) == "901" { + row["name"] = "conflicting same-time observation" + } + } + } + return rows, err +} + +func (f *isolatedWorkflowGitHub) GetWorkflowRun(ctx context.Context, owner, repo, runID string, reporter gh.Reporter) (map[string]any, error) { + if f.failure == "lookup" { + f.lookupCalls++ + return nil, errors.New("exact workflow lookup unavailable") + } + return f.sameSyncSharedHeadGitHub.GetWorkflowRun(ctx, owner, repo, runID, reporter) +} + +func TestSyncConsolidationFailureRetainsUnrelatedAcquisitions(t *testing.T) { + for _, failure := range []string{"conflict", "lookup"} { + for _, numbers := range [][]int{{7, 8, 9, 10}, {10, 9, 8, 7}} { + t.Run(fmt.Sprintf("%s/%v", failure, numbers), func(t *testing.T) { + ctx := context.Background() + st, err := store.Open(ctx, filepath.Join(t.TempDir(), "archive.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + client := &isolatedWorkflowGitHub{failure: failure} + opts := Options{Owner: "fixture", Repo: "repo", Numbers: numbers, IncludePRDetails: true} + stats, err := New(client, st).Sync(ctx, opts) + if err == nil || stats.ThreadsSynced != 2 || stats.PRDetailsSynced != 1 || stats.WorkflowRunsSynced != 1 { + t.Fatalf("unrelated acquisitions lost: %+v err=%v", stats, err) + } + repo, err := st.RepositoryByFullName(ctx, "fixture/repo") + if err != nil { + t.Fatal(err) + } + failures, err := st.ListSyncAttemptFailures(ctx, store.SyncAttemptFailureListOptions{RepoID: repo.ID}) + if err != nil || len(failures) != 2 { + t.Fatalf("affected group ledger=%+v err=%v", failures, err) + } + for _, failed := range failures { + if !slices.Contains([]int{8, 9}, failed.Number) || failed.Operation != "pull_request_details" || failed.ThreadID == 0 { + t.Fatalf("wrong group failure=%+v", failed) + } + } + runs, err := st.ListWorkflowRuns(ctx, repo.ID, store.WorkflowRunListOptions{Limit: -1}) + if err != nil || len(runs) != 1 || runs[0].HeadSHA != "other-head" { + t.Fatalf("failed shared head persisted=%+v err=%v", runs, err) + } + assertTableRowCount(t, st, "documents", 2) + assertNoSuccessfulSync(t, st, repo.ID) + opts.Numbers = []int{8, 9} + if _, err := New(&isolatedWorkflowGitHub{}, st).Sync(ctx, opts); err != nil { + t.Fatal(err) + } + failures, err = st.ListSyncAttemptFailures(ctx, store.SyncAttemptFailureListOptions{RepoID: repo.ID}) + if err != nil || len(failures) != 0 { + t.Fatalf("completed group retry unresolved=%+v err=%v", failures, err) + } + }) + } + } +} + +type quotaWorkflowLookupGitHub struct { + fakeGitHub + calls int +} + +func (f *quotaWorkflowLookupGitHub) GetWorkflowRun(context.Context, string, string, string, gh.Reporter) (map[string]any, error) { + f.calls++ + return nil, &gh.RateLimitReserveError{Reserve: 10, RateLimit: gh.RateLimitSnapshot{Resource: "core", Remaining: 10}} +} + +func TestConsolidationQuotaStopDoesNotAttemptOrBlameLaterGroups(t *testing.T) { + var payloads []threadSyncPayload + for group, head := range []string{"first", "later", "offline"} { + row := func(id int) map[string]any { + return map[string]any{"id": id, "head_sha": head, "created_at": "2026-07-12T00:00:00Z", "updated_at": "2026-07-12T00:01:00Z"} + } + for observation := 0; observation < 2; observation++ { + rows := []map[string]any{row(10*group + 1)} + if observation == 0 && head != "offline" { + rows = append(rows, row(10*group+2)) + } + payloads = append(payloads, threadSyncPayload{hasPullDetails: true, pullDetails: pullRequestDetailRows{ + pull: map[string]any{"head": map[string]any{"sha": head}}, runsRaw: rows, + workflowSnapshotFresh: true, workflowSourceUpdatedAt: "2026-07-12T00:01:00Z", + workflowObservationOrder: len(payloads) + 1, + }}) + } + } + client := "aWorkflowLookupGitHub{} + failures := New(client, nil).consolidateWorkflowSnapshots(context.Background(), Options{}, payloads, true) + if client.calls != 1 || len(failures) != 4 { + t.Fatalf("continued verification after quota: calls=%d failures=%v", client.calls, failures) + } + for _, index := range []int{0, 1} { + var reserveErr *gh.RateLimitReserveError + if !errors.As(failures[index], &reserveErr) { + t.Fatalf("actual group failure lost at %d: %v", index, failures[index]) + } + } + for _, index := range []int{2, 3} { + if cause, excluded := failures[index]; !excluded || cause != nil { + t.Fatalf("unattempted verification must be excluded without a failure: index=%d excluded=%v cause=%v", index, excluded, cause) + } + } + for _, index := range []int{4, 5} { + if _, excluded := failures[index]; excluded { + t.Fatalf("already complete offline group excluded: %d", index) + } + } +} + +func TestSyncNativeQuotaStopPreservesCompletedPayloadsWithoutFurtherRequests(t *testing.T) { + for _, phase := range []string{"parents", "comments"} { + t.Run(phase, func(t *testing.T) { + ctx := context.Background() + st, err := store.Open(ctx, filepath.Join(t.TempDir(), "archive.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + var requests []string + rateCalls, stopAt := 0, 3 + if phase == "comments" { + stopAt = 6 + } + resetAt := time.Now().Add(time.Hour).Unix() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests = append(requests, r.URL.Path) + if r.URL.Path == "/rate_limit" { + rateCalls++ + remaining := 100 + if rateCalls >= stopAt { + remaining = 10 + } + _ = json.NewEncoder(w).Encode(map[string]any{"resources": map[string]any{ + "core": map[string]any{"limit": 5000, "remaining": remaining, "reset": resetAt}, + }}) + return + } + switch { + case r.URL.Path == "/repos/fixture/repo": + _ = json.NewEncoder(w).Encode(map[string]any{"id": 123}) + case r.URL.Path == "/repos/fixture/repo/issues/7/comments": + _ = json.NewEncoder(w).Encode([]map[string]any{{"id": 11, "body": "retained", "updated_at": "2026-07-12T00:00:00Z"}}) + case strings.HasPrefix(r.URL.Path, "/repos/fixture/repo/issues/"): + number, err := strconv.Atoi(strings.TrimPrefix(r.URL.Path, "/repos/fixture/repo/issues/")) + if err != nil { + t.Errorf("unexpected request after quota stop: %s", r.URL.Path) + http.Error(w, "unexpected request", http.StatusBadRequest) + return + } + _ = json.NewEncoder(w).Encode(map[string]any{"id": number, "number": number, "state": "open", + "title": "fixture", "body": "retained", "created_at": "2026-07-12T00:00:00Z", "updated_at": "2026-07-12T00:00:00Z"}) + default: + t.Errorf("unexpected request: %s", r.URL.Path) + http.Error(w, "unexpected request", http.StatusBadRequest) + } + })) + defer server.Close() + client := gh.New(gh.Options{BaseURL: server.URL, RateLimitReserve: 10}) + stats, err := New(client, st).Sync(ctx, Options{Owner: "fixture", Repo: "repo", Numbers: []int{7, 8, 9}, IncludeComments: phase == "comments"}) + var reserveErr *gh.RateLimitReserveError + if !errors.As(err, &reserveErr) || reserveErr.Reserve != 10 || reserveErr.RateLimit.Remaining != 10 { + t.Fatalf("original quota error lost: %v", err) + } + if stats.ThreadsSynced != 1 || rateCalls != stopAt || len(requests) != 2*stopAt-1 || requests[len(requests)-1] != "/rate_limit" { + t.Fatalf("continued after quota stop: stats=%+v rate=%d requests=%v", stats, rateCalls, requests) + } + repo, err := st.RepositoryByFullName(ctx, "fixture/repo") + if err != nil { + t.Fatal(err) + } + failures, err := st.ListSyncAttemptFailures(ctx, store.SyncAttemptFailureListOptions{RepoID: repo.ID}) + operation := "issue" + if phase == "comments" { + operation = "issue_comments" + } + if err != nil || len(failures) != 1 || failures[0].Number != 8 || failures[0].Operation != operation { + t.Fatalf("unattempted item recorded=%+v err=%v", failures, err) + } + assertNoSuccessfulSync(t, st, repo.ID) + }) + } +} + +func TestSyncPersistenceFailureLedgerSurvivesRollbackAndRequiresObservedFamilies(t *testing.T) { + for _, fixture := range []struct { + number int + existing bool + }{{7, false}, {7, true}, {8, false}, {8, true}} { + t.Run(fmt.Sprint(fixture), func(t *testing.T) { + ctx := context.Background() + st, err := store.Open(ctx, filepath.Join(t.TempDir(), "archive.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + s := New(&partialGitHub{}, st) + opts := Options{Owner: "fixture", Repo: "repo", Numbers: []int{fixture.number}} + if fixture.existing { + if _, err := s.Sync(ctx, opts); err != nil { + t.Fatal(err) + } + } + if _, err := st.DB().ExecContext(ctx, `create trigger reject_fingerprint before insert on thread_fingerprints + begin select raise(abort, 'fingerprint rejected'); end`); err != nil { + t.Fatal(err) + } + opts.IncludeComments, opts.IncludePRDetails = true, true + stats, err := s.Sync(ctx, opts) + if err == nil || !strings.Contains(err.Error(), "fingerprint rejected") || stats.ThreadsSynced != 0 { + t.Fatalf("persistence failure=%+v err=%v", stats, err) + } + repo, err := st.RepositoryByFullName(ctx, "fixture/repo") + if err != nil { + t.Fatal(err) + } + failures, err := st.ListSyncAttemptFailures(ctx, store.SyncAttemptFailureListOptions{RepoID: repo.ID}) + operations := []string{"issue", "issue_comments"} + if fixture.number == 8 { + operations = append(operations, "pull_request_metadata", "pull_request_details", + "pull_review_threads", "pull_reviews", "pull_review_comments") + } + slices.Sort(operations) + assertFailureOperations := func(failures []store.SyncAttemptFailure, want []string) { + t.Helper() + var got []string + for _, failure := range failures { + got = append(got, failure.Operation) + } + slices.Sort(got) + if !slices.Equal(got, want) { + t.Fatalf("failure operations=%v want=%v", got, want) + } + } + if err != nil { + t.Fatalf("rollback failure ledger=%+v err=%v", failures, err) + } + assertFailureOperations(failures, operations) + for _, failure := range failures { + if (failure.ThreadID != 0) != fixture.existing || !strings.Contains(failure.ErrorMessage, "fingerprint rejected") { + t.Fatalf("rollback failure parent/cause=%+v", failure) + } + } + coverage, err := st.ArchiveCoverage(ctx, store.ArchiveCoverageOptions{}) + if err != nil || coverage.Totals.KnownFailedHydrations == nil || *coverage.Totals.KnownFailedHydrations != len(operations) { + t.Fatalf("inventory hides persistence failure: %+v err=%v", coverage.Totals, err) + } + assertTableRowCount(t, st, "thread_revisions", 0) + if _, err := s.Sync(ctx, opts); err == nil { + t.Fatal("failed retry unexpectedly succeeded") + } + failures, err = st.ListSyncAttemptFailures(ctx, store.SyncAttemptFailureListOptions{RepoID: repo.ID}) + if err != nil { + t.Fatalf("rolled-back retry cleared persistence failure=%+v err=%v", failures, err) + } + assertFailureOperations(failures, operations) + for _, failure := range failures { + if failure.RetryCount != 1 || failure.ResolvedAt != "" { + t.Fatalf("rolled-back retry resolved or failed to count=%+v", failure) + } + } + shallow := opts + shallow.IncludeComments, shallow.IncludePRDetails, shallow.IncludePRMetadata = false, false, true + if _, err := s.Sync(ctx, shallow); err != nil { + t.Fatal(err) + } + failures, err = st.ListSyncAttemptFailures(ctx, store.SyncAttemptFailureListOptions{RepoID: repo.ID}) + if err != nil { + t.Fatal(err) + } + remaining := []string{"issue_comments"} + if fixture.number == 8 { + remaining = append(remaining, "pull_request_details", "pull_review_comments", "pull_review_threads", "pull_reviews") + } + assertFailureOperations(failures, remaining) + if _, err := st.DB().ExecContext(ctx, `drop trigger reject_fingerprint`); err != nil { + t.Fatal(err) + } + shallow.IncludeComments, shallow.IncludePRMetadata = true, false + if _, err := s.Sync(ctx, shallow); err != nil { + t.Fatal(err) + } + failures, err = st.ListSyncAttemptFailures(ctx, store.SyncAttemptFailureListOptions{RepoID: repo.ID}) + if err != nil { + t.Fatal(err) + } + remaining = nil + if fixture.number == 8 { + remaining = []string{"pull_request_details", "pull_review_threads"} + } + assertFailureOperations(failures, remaining) + if _, err := s.Sync(ctx, opts); err != nil { + t.Fatal(err) + } + failures, err = st.ListSyncAttemptFailures(ctx, store.SyncAttemptFailureListOptions{RepoID: repo.ID, IncludeResolved: true}) + if err != nil { + t.Fatalf("complete retry failed to resolve=%+v err=%v", failures, err) + } + assertFailureOperations(failures, operations) + for _, failure := range failures { + if failure.ResolvedAt == "" { + t.Fatalf("complete retry left unresolved family=%+v", failure) + } + } + }) + } +} + +func TestSyncPersistenceBookkeepingFailurePreservesOriginalErrorAndPriorCommit(t *testing.T) { + ctx := context.Background() + st, err := store.Open(ctx, filepath.Join(t.TempDir(), "archive.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + if _, err := st.DB().ExecContext(ctx, ` + create trigger reject_second before insert on documents + when (select number from threads where id = new.thread_id) = 8 + begin select raise(abort, 'original persistence failure'); end; + create trigger reject_failure_record before insert on sync_attempt_failures + when new.operation = 'issue_comments' + begin select raise(abort, 'failure ledger unavailable'); end;`); err != nil { + t.Fatal(err) + } + stats, err := New(&partialGitHub{}, st).Sync(ctx, Options{ + Owner: "fixture", Repo: "repo", Numbers: []int{7, 8, 9}, IncludeComments: true, + }) + if err == nil || !strings.Contains(err.Error(), "original persistence failure") || + !strings.Contains(err.Error(), "failure ledger unavailable") || stats.ThreadsSynced != 1 { + t.Fatalf("original failure or prior commit lost: %+v err=%v", stats, err) + } + assertTableRowCount(t, st, "threads", 1) + assertTableRowCount(t, st, "documents", 1) + assertTableRowCount(t, st, "sync_attempt_failures", 0) +} + +func TestSyncChildFailureResolvesOnlyCommittedParentFailure(t *testing.T) { + for _, fixture := range []struct { + number int + operation string + guard string + }{ + {7, "issue_comments", ""}, + {8, "issue_comments", ""}, + {8, "pull_reviews", ""}, + {8, "pull_review_comments", ""}, + {8, "pull_review_threads", ""}, + {8, "pull_request_metadata", ""}, + {8, "pull_request_details", ""}, + {8, "issue_comments", "stale"}, + {8, "issue_comments", "failure rollback"}, + {8, "issue_comments", "resolution rollback"}, + } { + t.Run(fmt.Sprint(fixture), func(t *testing.T) { + ctx := context.Background() + st, err := store.Open(ctx, filepath.Join(t.TempDir(), "archive.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + client := &partialGitHub{failNumber: fixture.number, operation: "issue"} + s := New(client, st) + now := time.Date(2026, 7, 16, 0, 0, 0, 0, time.UTC) + s.now = func() time.Time { return now } + opts := Options{Owner: "fixture", Repo: "repo", Numbers: []int{fixture.number}, + IncludeComments: true, IncludePRMetadata: true, + IncludePRDetails: fixture.operation != "pull_request_metadata"} + if _, err := s.Sync(ctx, opts); err == nil { + t.Fatal("initial parent fetch unexpectedly succeeded") + } + repo, err := st.RepositoryByFullName(ctx, "fixture/repo") + if err != nil { + t.Fatal(err) + } + watermark, err := st.ClosedSweepWatermark(ctx, repo.ID) + if err != nil { + t.Fatal(err) + } + client.operation = fixture.operation + switch fixture.guard { + case "stale": + row, err := client.GetIssue(ctx, opts.Owner, opts.Repo, fixture.number, nil) + if err != nil { + t.Fatal(err) + } + row["title"], row["updated_at"] = "newer retained parent", "2026-07-15T00:00:00Z" + if _, err := st.UpsertThread(ctx, mapIssueToThread(repo.ID, row, now.Format(time.RFC3339Nano)), + store.UpsertThreadOptions{IncompleteEvidence: true}); err != nil { + t.Fatal(err) + } + case "failure rollback": + if _, err := st.DB().ExecContext(ctx, `create trigger reject_child_failure before insert on sync_attempt_failures + when new.operation = 'issue_comments' + begin select raise(abort, 'failure ledger rejected'); end`); err != nil { + t.Fatal(err) + } + case "resolution rollback": + if _, err := st.DB().ExecContext(ctx, `create trigger reject_parent_resolution before update on sync_attempt_failures + when new.operation = 'issue' and new.resolved_at is not null + begin select raise(abort, 'parent resolution rejected'); end`); err != nil { + t.Fatal(err) + } + } + now = now.Add(time.Hour) + stats, err := s.Sync(ctx, opts) + if err == nil || !strings.Contains(err.Error(), fixture.operation+" unavailable") || + stats.ThreadsSynced != 0 || stats.EvidenceObserved != 0 || stats.ClosedSweepThrough != "" { + t.Fatalf("child failure or incomplete counts lost: %+v err=%v", stats, err) + } + rollback := strings.HasSuffix(fixture.guard, "rollback") + if rollback && !strings.Contains(err.Error(), "rejected") { + t.Fatalf("bookkeeping failure lost: %v", err) + } + failures, err := st.ListSyncAttemptFailures(ctx, store.SyncAttemptFailureListOptions{RepoID: repo.ID, IncludeResolved: true}) + wantFailures := 2 + if rollback { + wantFailures = 1 + } + if err != nil || len(failures) != wantFailures { + t.Fatalf("failure history=%+v err=%v", failures, err) + } + for _, failure := range failures { + switch failure.Operation { + case "issue": + if (failure.ResolvedAt != "") != (fixture.guard == "") { + t.Fatalf("parent resolution disagrees with committed observation: %+v", failure) + } + case fixture.operation: + if failure.ResolvedAt != "" || failure.ThreadID == 0 { + t.Fatalf("failed child incorrectly resolved or detached: %+v", failure) + } + default: + t.Fatalf("unexpected failure family: %+v", failure) + } + } + wantThreads := 1 + if rollback { + wantThreads = 0 + } + assertTableRowCount(t, st, "threads", wantThreads) + if wantThreads > 0 { + var title, evidenceAt string + var sequence, evidenceSequence int64 + if err := st.DB().QueryRowContext(ctx, `select title, observation_sequence, + evidence_source_updated_at, evidence_observation_sequence from threads`).Scan( + &title, &sequence, &evidenceAt, &evidenceSequence); err != nil { + t.Fatal(err) + } + if sequence >= 0 || evidenceSequence != 0 || evidenceAt != "" || + fixture.guard == "stale" && title != "newer retained parent" { + t.Fatalf("parent freshness overclaimed: title=%q sequence=%d evidence=%d/%q", title, sequence, evidenceSequence, evidenceAt) + } + } + for _, table := range []string{"comments", "documents", "thread_revisions", "thread_fingerprints", "thread_child_observation_reservations"} { + assertTableRowCount(t, st, table, 0) + } + assertNoSuccessfulSync(t, st, repo.ID) + after, err := st.ClosedSweepWatermark(ctx, repo.ID) + if err != nil || !after.Equal(watermark) { + t.Fatalf("failed retry advanced checkpoint: before=%v after=%v err=%v", watermark, after, err) + } + }) + } +} + +func TestSyncNoPersistedFamilyDoesNotResolveFailures(t *testing.T) { + ctx := context.Background() + st, err := store.Open(ctx, filepath.Join(t.TempDir(), "archive.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + client := &partialGitHub{} + s := New(client, st) + opts := Options{Owner: "fixture", Repo: "repo", Numbers: []int{7}, IncludeComments: true} + if _, err := s.Sync(ctx, opts); err != nil { + t.Fatal(err) + } + repo, err := st.RepositoryByFullName(ctx, "fixture/repo") + if err != nil { + t.Fatal(err) + } + if _, err := st.RecordSyncAttemptFailure(ctx, store.SyncAttemptFailure{ + RepoID: repo.ID, Number: 7, Operation: "issue_comments", ErrorMessage: "unavailable", + LastSeenAt: time.Now().UTC().Format(time.RFC3339Nano), + }); err != nil { + t.Fatal(err) + } + s.beforePersist = func() { + row, err := client.GetIssue(ctx, opts.Owner, opts.Repo, 7, nil) + if err != nil { + t.Fatal(err) + } + thread := mapIssueToThread(repo.ID, row, time.Now().UTC().Format(time.RFC3339Nano)) + if err := st.WithTx(ctx, func(tx *store.Store) error { + sequence, err := tx.NextThreadObservationSequence(ctx, thread.UpdatedAt) + if err != nil { + return err + } + // A competing observation owns the parent and comments, but has not + // advanced complete evidence. This payload can persist only enrichment. + upsert, err := tx.UpsertThreadObservation(ctx, thread, store.UpsertThreadOptions{ + IncompleteEvidence: true, ObservationSequence: sequence, + }) + if err != nil { + return err + } + _, err = tx.ReserveThreadChildObservation(ctx, upsert.ID, store.ThreadChildComments, thread.UpdatedAtGitHub, sequence) + return err + }); err != nil { + t.Fatal(err) + } + } + stats, err := s.Sync(ctx, opts) + if err != nil || stats.ThreadsSynced != 1 || stats.CommentsSynced != 0 || stats.EvidenceObserved != 1 { + t.Fatalf("expected enrichment without parent/child replacement: %+v err=%v", stats, err) + } + failures, err := st.ListSyncAttemptFailures(ctx, store.SyncAttemptFailureListOptions{RepoID: repo.ID}) + if err != nil || len(failures) != 1 || failures[0].Operation != "issue_comments" { + t.Fatalf("empty family set resolved failures: %+v err=%v", failures, err) + } +} diff --git a/internal/syncer/partial_sync_test.go b/internal/syncer/partial_sync_test.go new file mode 100644 index 0000000..8ebb36c --- /dev/null +++ b/internal/syncer/partial_sync_test.go @@ -0,0 +1,484 @@ +package syncer + +import ( + "context" + "database/sql" + "errors" + "fmt" + "log/slog" + "path/filepath" + "slices" + "strings" + "testing" + "time" + + gh "github.com/openclaw/gitcrawl/internal/github" + "github.com/openclaw/gitcrawl/internal/store" +) + +type partialGitHub struct { + fakeGitHub + failNumber int + operation string + calls []int + cancel context.CancelFunc +} + +func (f *partialGitHub) fail(number int, operation string) error { + if number != f.failNumber || operation != f.operation { + return nil + } + if f.cancel != nil { + f.cancel() + return context.Canceled + } + return fmt.Errorf("%s unavailable", operation) +} + +func (f *partialGitHub) GetIssue(ctx context.Context, owner, repo string, number int, reporter gh.Reporter) (map[string]any, error) { + f.calls = append(f.calls, number) + if err := f.fail(number, "issue"); err != nil { + return nil, err + } + row, err := f.fakeGitHub.GetIssue(ctx, owner, repo, number, reporter) + row["id"], row["number"] = number, number + return row, err +} + +func (f *partialGitHub) ListIssueComments(ctx context.Context, owner, repo string, number int, reporter gh.Reporter) ([]map[string]any, error) { + if err := f.fail(number, "issue_comments"); err != nil { + return nil, err + } + return f.fakeGitHub.ListIssueComments(ctx, owner, repo, number, reporter) +} + +func (f *partialGitHub) ListPullReviews(ctx context.Context, owner, repo string, number int, reporter gh.Reporter) ([]map[string]any, error) { + if err := f.fail(number, "pull_reviews"); err != nil { + return nil, err + } + return f.fakeGitHub.ListPullReviews(ctx, owner, repo, number, reporter) +} + +func (f *partialGitHub) ListPullReviewComments(ctx context.Context, owner, repo string, number int, reporter gh.Reporter) ([]map[string]any, error) { + if err := f.fail(number, "pull_review_comments"); err != nil { + return nil, err + } + return f.fakeGitHub.ListPullReviewComments(ctx, owner, repo, number, reporter) +} + +func (f *partialGitHub) ListPullReviewThreads(ctx context.Context, owner, repo string, number int, reporter gh.Reporter) ([]map[string]any, error) { + if err := f.fail(number, "pull_review_threads"); err != nil { + return nil, err + } + return f.fakeGitHub.ListPullReviewThreads(ctx, owner, repo, number, reporter) +} + +func (f *partialGitHub) GetPull(ctx context.Context, owner, repo string, number int, reporter gh.Reporter) (map[string]any, error) { + if err := f.fail(number, "pull_request_metadata"); err != nil { + return nil, err + } + return f.fakeGitHub.GetPull(ctx, owner, repo, number, reporter) +} + +func (f *partialGitHub) ListPullFiles(ctx context.Context, owner, repo string, number int, reporter gh.Reporter) ([]map[string]any, error) { + if err := f.fail(number, "pull_request_details"); err != nil { + return nil, err + } + return f.fakeGitHub.ListPullFiles(ctx, owner, repo, number, reporter) +} + +func TestSyncPartialItemFailuresRetainHealthySiblingsAndRetry(t *testing.T) { + for _, operation := range []string{ + "issue", "issue_comments", "pull_reviews", "pull_review_comments", + "pull_review_threads", "pull_request_metadata", "pull_request_details", + } { + for _, numbers := range [][]int{{8, 7, 9}, {7, 8, 9}, {7, 9, 8}} { + t.Run(fmt.Sprintf("%s/%v", operation, numbers), func(t *testing.T) { + ctx := context.Background() + st, err := store.Open(ctx, filepath.Join(t.TempDir(), "archive.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + client := &partialGitHub{operation: operation, failNumber: 8} + s := New(client, st) + opts := Options{Owner: "fixture", Repo: "repo", Numbers: numbers, IncludeComments: true, + IncludePRDetails: operation != "pull_request_metadata", IncludePRMetadata: true} + stats, err := s.Sync(ctx, opts) + if err == nil || !strings.Contains(err.Error(), operation+" unavailable") { + t.Fatalf("failure=%v", err) + } + if stats.ThreadsSynced != 2 || stats.IssuesSynced != 2 || stats.CommentsSynced != 1 || + stats.EvidenceObserved != 2 || stats.ClosedSweepThrough != "" { + t.Fatalf("committed counts=%+v", stats) + } + repo, err := st.RepositoryByFullName(ctx, "fixture/repo") + if err != nil { + t.Fatal(err) + } + failures, err := st.ListSyncAttemptFailures(ctx, store.SyncAttemptFailureListOptions{RepoID: repo.ID}) + if err != nil || len(failures) != 1 || failures[0].Operation != operation || failures[0].Number != 8 { + t.Fatalf("failure ledger=%+v err=%v", failures, err) + } + if (failures[0].ThreadID == 0) != (operation == "issue") { + t.Fatalf("only an observed parent can own a thread ID: %+v", failures[0]) + } + assertTableRowCount(t, st, "documents", 2) + assertTableRowCount(t, st, "thread_revisions", 2) + assertNoSuccessfulSync(t, st, repo.ID) + client.operation = "" + opts.Numbers = []int{8} + if _, err := s.Sync(ctx, opts); err != nil { + t.Fatal(err) + } + failures, err = st.ListSyncAttemptFailures(ctx, store.SyncAttemptFailureListOptions{RepoID: repo.ID}) + if err != nil || len(failures) != 0 { + t.Fatalf("successful retry unresolved=%+v err=%v", failures, err) + } + }) + } + } +} + +func assertNoSuccessfulSync(t *testing.T, st *store.Store, repoID int64) { + t.Helper() + ctx := context.Background() + last, err := st.LastSuccessfulSyncAt(ctx, repoID) + if err != nil || !last.IsZero() { + t.Fatalf("partial run certified sync freshness=%v err=%v", last, err) + } + last, err = st.LastSuccessfulListSyncAt(ctx, repoID, "open") + if err != nil || !last.IsZero() { + t.Fatalf("partial run certified list freshness=%v err=%v", last, err) + } + runs, err := st.SuccessfulListSyncRuns(ctx, repoID, "open") + if err != nil || len(runs) != 0 { + t.Fatalf("partial run certified coverage=%+v err=%v", runs, err) + } +} + +func TestSyncPartialPersistenceRollsBackOnlyFailedItem(t *testing.T) { + ctx := context.Background() + st, err := store.Open(ctx, filepath.Join(t.TempDir(), "archive.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + if _, err := st.DB().ExecContext(ctx, ` + create trigger reject_middle_fingerprint before insert on thread_fingerprints + when (select t.number from threads t join thread_revisions r on r.thread_id = t.id where r.id = new.thread_revision_id) = 8 + begin select raise(abort, 'middle fingerprint rejected'); end`); err != nil { + t.Fatal(err) + } + stats, err := New(&partialGitHub{}, st).Sync(ctx, Options{ + Owner: "fixture", Repo: "repo", Numbers: []int{7, 8, 9}, IncludeComments: true, IncludePRDetails: true, + }) + if err == nil || !strings.Contains(err.Error(), "middle fingerprint rejected") { + t.Fatalf("failure=%v", err) + } + if stats.ThreadsSynced != 2 || stats.EvidenceObserved != 2 || stats.RevisionsCreated != 2 || + stats.FingerprintsUpserted != 2 || stats.PRDetailsSynced != 0 { + t.Fatalf("rolled-back item counted=%+v", stats) + } + for _, table := range []string{"threads", "documents", "thread_revisions", "thread_fingerprints"} { + assertTableRowCount(t, st, table, 2) + } + for _, table := range []string{"pull_request_details", "pull_request_review_thread_syncs"} { + assertTableRowCount(t, st, table, 0) + } + var reservations int + if err := st.DB().QueryRowContext(ctx, `select count(*) from thread_child_observation_reservations where thread_id not in (select id from threads)`).Scan(&reservations); err != nil || reservations != 0 { + t.Fatalf("orphan child reservations=%d err=%v", reservations, err) + } + repo, err := st.RepositoryByFullName(ctx, "fixture/repo") + if err != nil { + t.Fatal(err) + } + assertNoSuccessfulSync(t, st, repo.ID) +} + +func TestSyncFailureResolutionRequiresObservedFamilies(t *testing.T) { + ctx := context.Background() + st, err := store.Open(ctx, filepath.Join(t.TempDir(), "archive.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + client := &partialGitHub{} + s := New(client, st) + opts := Options{Owner: "fixture", Repo: "repo", Numbers: []int{8}} + if _, err := s.Sync(ctx, opts); err != nil { + t.Fatal(err) + } + repo, err := st.RepositoryByFullName(ctx, "fixture/repo") + if err != nil { + t.Fatal(err) + } + operations := []string{"issue", "issue_comments", "pull_reviews", "pull_review_comments", "pull_review_threads", "pull_request_metadata", "pull_request_details"} + for _, operation := range operations { + if _, err := st.RecordSyncAttemptFailure(ctx, store.SyncAttemptFailure{ + RepoID: repo.ID, Number: 8, Operation: operation, ErrorMessage: "unavailable", LastSeenAt: time.Now().UTC().Format(time.RFC3339Nano), + }); err != nil { + t.Fatal(err) + } + } + opts.IncludePRDetails = true + if _, err := s.Sync(ctx, opts); err != nil { + t.Fatal(err) + } + failures, err := st.ListSyncAttemptFailures(ctx, store.SyncAttemptFailureListOptions{RepoID: repo.ID}) + if err != nil { + t.Fatal(err) + } + var remaining []string + for _, failure := range failures { + remaining = append(remaining, failure.Operation) + } + slices.Sort(remaining) + if !slices.Equal(remaining, []string{"issue_comments", "pull_review_comments", "pull_reviews"}) { + t.Fatalf("unobserved comments resolved=%v", remaining) + } + opts.IncludePRDetails, opts.IncludeComments = false, true + if _, err := s.Sync(ctx, opts); err != nil { + t.Fatal(err) + } + failures, err = st.ListSyncAttemptFailures(ctx, store.SyncAttemptFailureListOptions{RepoID: repo.ID}) + if err != nil || len(failures) != 0 { + t.Fatalf("observed comments unresolved=%+v err=%v", failures, err) + } +} + +func TestSyncCancellationRecordsFailureAndStopsAcquisition(t *testing.T) { + background := context.Background() + ctx, cancel := context.WithCancel(background) + defer cancel() + st, err := store.Open(background, filepath.Join(t.TempDir(), "archive.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + client := &partialGitHub{failNumber: 8, operation: "issue", cancel: cancel} + _, err = New(client, st).Sync(ctx, Options{Owner: "fixture", Repo: "repo", Numbers: []int{7, 8, 9}}) + if !errors.Is(err, context.Canceled) || !slices.Equal(client.calls, []int{7, 8}) { + t.Fatalf("cancellation=%v calls=%v", err, client.calls) + } + repo, err := st.RepositoryByFullName(background, "fixture/repo") + if err != nil { + t.Fatal(err) + } + failures, err := st.ListSyncAttemptFailures(background, store.SyncAttemptFailureListOptions{RepoID: repo.ID}) + if err != nil || len(failures) != 1 || failures[0].ErrorClass != "context_canceled" { + t.Fatalf("cancel failure=%+v err=%v", failures, err) + } + assertNoSuccessfulSync(t, st, repo.ID) +} + +type cancelAfterCommitWriter struct{ cancel context.CancelFunc } + +func (w cancelAfterCommitWriter) Write(data []byte) (int, error) { + if strings.Contains(string(data), "state=progress done=1 ") { + w.cancel() + } + return len(data), nil +} + +func TestSyncCancellationRetainsOnlyCommittedItems(t *testing.T) { + for _, afterCommit := range []bool{false, true} { + t.Run(fmt.Sprint(afterCommit), func(t *testing.T) { + background := context.Background() + ctx, cancel := context.WithCancel(background) + defer cancel() + st, err := store.Open(background, filepath.Join(t.TempDir(), "archive.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + s := New(&partialGitHub{}, st) + opts := Options{Owner: "fixture", Repo: "repo", Numbers: []int{7, 9}, IncludeComments: true} + want := 0 + if afterCommit { + want = 1 + opts.Logger = slog.New(slog.NewTextHandler(cancelAfterCommitWriter{cancel}, nil)) + } else { + s.beforePersist = func() { + calls := 0 + s.now = func() time.Time { + calls++ + // Repository and parent timestamps precede enrichment. + // Cancel after the parent and comments were written. + if calls == 3 { + cancel() + } + return time.Now().UTC() + } + } + } + stats, err := s.Sync(ctx, opts) + if !errors.Is(err, context.Canceled) || stats.ThreadsSynced != want || stats.EvidenceObserved != want { + t.Fatalf("cancellation counts=%+v err=%v", stats, err) + } + for _, table := range []string{"threads", "comments", "documents", "thread_revisions", "thread_fingerprints"} { + assertTableRowCount(t, st, table, want) + } + }) + } +} + +func TestSyncBusyCommitRetryDoesNotDoubleCount(t *testing.T) { + ctx := context.Background() + path := filepath.Join(t.TempDir(), "archive.db") + st, err := store.Open(ctx, path) + if err != nil { + t.Fatal(err) + } + defer st.Close() + // A rollback-journal reader blocks COMMIT after every item write succeeded. + // This exercises the driver's rollback and WithTx's complete callback retry. + if _, err := st.DB().ExecContext(ctx, `pragma journal_mode = delete; pragma busy_timeout = 1`); err != nil { + t.Fatal(err) + } + reader, err := sql.Open("sqlite", path) + if err != nil { + t.Fatal(err) + } + defer reader.Close() + s := New(&partialGitHub{}, st) + var readTx *sql.Tx + calls := 0 + s.beforePersist = func() { + readTx, err = reader.BeginTx(ctx, nil) + if err != nil { + t.Fatal(err) + } + var count int + if err := readTx.QueryRowContext(ctx, `select count(*) from repositories`).Scan(&count); err != nil { + t.Fatal(err) + } + s.now = func() time.Time { + calls++ + // Four reads cover repository, parent, enrichment, and resolution. + // The fifth starts the retried item after the first COMMIT was busy. + if calls == 5 { + if err := readTx.Rollback(); err != nil { + t.Fatal(err) + } + } + return time.Now().UTC() + } + } + defer func() { + if readTx != nil { + _ = readTx.Rollback() + } + }() + stats, err := s.Sync(ctx, Options{Owner: "fixture", Repo: "repo", Numbers: []int{7}, IncludeComments: true}) + if err != nil { + t.Fatal(err) + } + if calls < 9 || stats.ThreadsSynced != 1 || stats.CommentsSynced != 1 || + stats.EvidenceObserved != 1 || stats.RevisionsCreated != 1 || stats.FingerprintsUpserted != 1 { + t.Fatalf("retry counts=%+v clock reads=%d", stats, calls) + } + for _, table := range []string{"threads", "comments", "documents", "thread_revisions", "thread_fingerprints"} { + assertTableRowCount(t, st, table, 1) + } +} + +type partialSharedHeadGitHub struct{ *sameSyncSharedHeadGitHub } + +func (f partialSharedHeadGitHub) GetIssue(ctx context.Context, owner, repo string, number int, reporter gh.Reporter) (map[string]any, error) { + if number == 10 { + return nil, errors.New("issue unavailable") + } + return f.sameSyncSharedHeadGitHub.GetIssue(ctx, owner, repo, number, reporter) +} + +func TestSyncPartialBatchPreservesSharedHeadConsolidation(t *testing.T) { + for _, fixture := range []struct { + name string + subsetFirst bool + verifiedDeletion bool + wantIDs []string + }{ + {name: "subset first", subsetFirst: true, wantIDs: []string{"900", "901"}}, + {name: "superset first", wantIDs: []string{"900", "901"}}, + {name: "later deletion", verifiedDeletion: true, wantIDs: []string{"900"}}, + } { + for _, numbers := range [][]int{{10, 8, 9}, {8, 10, 9}, {8, 9, 10}} { + t.Run(fmt.Sprintf("%s/%v", fixture.name, numbers), func(t *testing.T) { + ctx := context.Background() + st, err := store.Open(ctx, filepath.Join(t.TempDir(), "archive.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + client := sameSyncSharedHeadGitHub{ + subsetFirst: fixture.subsetFirst, verifiedDeletion: fixture.verifiedDeletion, + } + stats, err := New(partialSharedHeadGitHub{&client}, st).Sync(ctx, Options{ + Owner: "fixture", Repo: "repo", Numbers: numbers, IncludePRDetails: true, + }) + if err == nil || stats.PRDetailsSynced != 2 || stats.WorkflowRunsSynced != len(fixture.wantIDs) { + t.Fatalf("partial shared-head stats=%+v err=%v", stats, err) + } + repo, err := st.RepositoryByFullName(ctx, "fixture/repo") + if err != nil { + t.Fatal(err) + } + runs, err := st.ListWorkflowRuns(ctx, repo.ID, store.WorkflowRunListOptions{HeadSHA: "same-sync-head", Limit: -1}) + if err != nil { + t.Fatal(err) + } + var ids []string + for _, run := range runs { + ids = append(ids, run.RunID) + } + slices.Sort(ids) + if !slices.Equal(ids, fixture.wantIDs) { + t.Fatalf("shared-head rows=%v want=%v", ids, fixture.wantIDs) + } + assertWorkflowRunReservation(t, ctx, st, repo.ID, "same-sync-head", 1) + assertNoSuccessfulSync(t, st, repo.ID) + }) + } + } +} + +func TestSyncSharedHeadSurvivesSiblingTransactionFailure(t *testing.T) { + for _, failedNumber := range []int{8, 9} { + t.Run(fmt.Sprint(failedNumber), func(t *testing.T) { + ctx := context.Background() + st, err := store.Open(ctx, filepath.Join(t.TempDir(), "archive.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + if _, err := st.DB().ExecContext(ctx, fmt.Sprintf(` + create trigger reject_sibling_document before insert on documents + when (select number from threads where id = new.thread_id) = %d + begin select raise(abort, 'sibling document rejected'); end`, failedNumber)); err != nil { + t.Fatal(err) + } + stats, err := New(&sameSyncSharedHeadGitHub{}, st).Sync(ctx, Options{ + Owner: "fixture", Repo: "repo", Numbers: []int{8, 9}, IncludePRDetails: true, + }) + if err == nil || stats.PRDetailsSynced != 1 || stats.WorkflowRunsSynced != 2 { + t.Fatalf("shared-head transaction counts=%+v err=%v", stats, err) + } + repo, err := st.RepositoryByFullName(ctx, "fixture/repo") + if err != nil { + t.Fatal(err) + } + threads, err := st.ListThreads(ctx, repo.ID, true) + if err != nil || len(threads) != 1 || threads[0].Number == failedNumber { + t.Fatalf("failed sibling retained=%+v err=%v", threads, err) + } + runs, err := st.ListWorkflowRuns(ctx, repo.ID, store.WorkflowRunListOptions{HeadSHA: "same-sync-head", Limit: -1}) + if err != nil || len(runs) != 2 { + t.Fatalf("consolidated workflows lost=%+v err=%v", runs, err) + } + assertWorkflowRunReservation(t, ctx, st, repo.ID, "same-sync-head", 1) + assertNoSuccessfulSync(t, st, repo.ID) + }) + } +} diff --git a/internal/syncer/pull_details.go b/internal/syncer/pull_details.go index 71089b9..13df61f 100644 --- a/internal/syncer/pull_details.go +++ b/internal/syncer/pull_details.go @@ -266,11 +266,16 @@ func (s *Syncer) workflowSnapshotObservation( return sourceUpdatedAt, true, baseline, deletedRunIDs, nil } +var errWorkflowVerificationNotAttempted = errors.New("workflow verification not attempted after acquisition stopped") + +// A present index excludes that payload. A nil error means verification was +// deliberately not attempted, so it must not create an acquisition failure. func (s *Syncer) consolidateWorkflowSnapshots( ctx context.Context, options Options, payloads []threadSyncPayload, -) error { + allowRequests bool, +) map[int]error { type snapshotGroup struct { headSHA string baseline store.WorkflowRunSnapshotState @@ -309,108 +314,91 @@ func (s *Syncer) consolidateWorkflowSnapshots( groups[groupIndex].indices = append(groups[groupIndex].indices, index) } + var failures map[int]error for _, group := range groups { if len(group.indices) < 2 { continue } - observationIndices := append([]int(nil), group.indices...) - sort.SliceStable(observationIndices, func(i, j int) bool { - left := payloads[observationIndices[i]].pullDetails.workflowObservationOrder - right := payloads[observationIndices[j]].pullDetails.workflowObservationOrder - if left <= 0 { - left = observationIndices[i] + 1 - } - if right <= 0 { - right = observationIndices[j] + 1 - } - return left < right - }) - sourceUpdatedAt := "" - rowsByID := make(map[string]map[string]any) - deletedRunIDs := make(map[string]struct{}) - observedRunIDs := make(map[string]struct{}) - laterAbsentRunIDs := make(map[string]struct{}) - for _, index := range observationIndices { - rows := &payloads[index].pullDetails - var err error - sourceUpdatedAt, err = latestWorkflowTimestamp( - sourceUpdatedAt, - rows.workflowSourceUpdatedAt, - ) - if err != nil { - return err - } - for _, runID := range rows.workflowDeletedRunIDs { - if runID != "" { - deletedRunIDs[runID] = struct{}{} - } - } - presentRunIDs := make(map[string]struct{}, len(rows.runsRaw)) - for _, row := range rows.runsRaw { - runID := jsonID(row["id"]) - if runID == "" { - continue - } - presentRunIDs[runID] = struct{}{} - existing, found := rowsByID[runID] - if !found { - rowsByID[runID] = row - continue - } - incomingSource, err := workflowRunTimestamp( - stringValue(row["updated_at"]), - stringValue(row["created_at"]), - ) - if err != nil { - return fmt.Errorf("workflow run %s source: %w", runID, err) - } - existingSource, err := workflowRunTimestamp( - stringValue(existing["updated_at"]), - stringValue(existing["created_at"]), - ) - if err != nil { - return fmt.Errorf("workflow run %s source: %w", runID, err) - } - switch { - case workflowTimestampBefore(existingSource, incomingSource): - rowsByID[runID] = row - case workflowTimestampBefore(incomingSource, existingSource): - default: - if mustJSON(existing) != mustJSON(row) { - return fmt.Errorf( - "conflicting workflow run %s observations share one sync generation", - runID, - ) - } - } - } - for runID := range observedRunIDs { - if _, present := presentRunIDs[runID]; !present { - laterAbsentRunIDs[runID] = struct{}{} - } - } - for runID := range presentRunIDs { - observedRunIDs[runID] = struct{}{} + err := s.consolidateWorkflowSnapshotGroup(ctx, options, payloads, group.indices, group.headSHA, allowRequests) + if err == nil { + continue + } + if failures == nil { + failures = make(map[int]error) + } + var reserveErr *gh.RateLimitReserveError + if errors.As(err, &reserveErr) || ctx.Err() != nil { + allowRequests = false + } + if errors.Is(err, errWorkflowVerificationNotAttempted) { + err = nil + } + for _, index := range group.indices { + failures[index] = err + } + } + return failures +} + +func (s *Syncer) consolidateWorkflowSnapshotGroup( + ctx context.Context, + options Options, + payloads []threadSyncPayload, + indices []int, + headSHA string, + allowRequests bool, +) error { + observationIndices := append([]int(nil), indices...) + sort.SliceStable(observationIndices, func(i, j int) bool { + left := payloads[observationIndices[i]].pullDetails.workflowObservationOrder + right := payloads[observationIndices[j]].pullDetails.workflowObservationOrder + if left <= 0 { + left = observationIndices[i] + 1 + } + if right <= 0 { + right = observationIndices[j] + 1 + } + return left < right + }) + sourceUpdatedAt := "" + rowsByID := make(map[string]map[string]any) + deletedRunIDs := make(map[string]struct{}) + observedRunIDs := make(map[string]struct{}) + laterAbsentRunIDs := make(map[string]struct{}) + for _, index := range observationIndices { + rows := &payloads[index].pullDetails + var err error + sourceUpdatedAt, err = latestWorkflowTimestamp( + sourceUpdatedAt, + rows.workflowSourceUpdatedAt, + ) + if err != nil { + return err + } + for _, runID := range rows.workflowDeletedRunIDs { + if runID != "" { + deletedRunIDs[runID] = struct{}{} } } - for runID := range laterAbsentRunIDs { - if _, deleted := deletedRunIDs[runID]; deleted { + presentRunIDs := make(map[string]struct{}, len(rows.runsRaw)) + for _, row := range rows.runsRaw { + runID := jsonID(row["id"]) + if runID == "" { continue } - exact, deleted, err := s.verifySiblingWorkflowRunDeletion( - ctx, - options, - group.headSHA, - runID, + presentRunIDs[runID] = struct{}{} + existing, found := rowsByID[runID] + if !found { + rowsByID[runID] = row + continue + } + incomingSource, err := workflowRunTimestamp( + stringValue(row["updated_at"]), + stringValue(row["created_at"]), ) if err != nil { - return err - } - if deleted { - deletedRunIDs[runID] = struct{}{} - continue + return fmt.Errorf("workflow run %s source: %w", runID, err) } - existing := rowsByID[runID] existingSource, err := workflowRunTimestamp( stringValue(existing["updated_at"]), stringValue(existing["created_at"]), @@ -418,52 +406,101 @@ func (s *Syncer) consolidateWorkflowSnapshots( if err != nil { return fmt.Errorf("workflow run %s source: %w", runID, err) } - exactSource, err := workflowRunTimestamp( - stringValue(exact["updated_at"]), - stringValue(exact["created_at"]), - ) - if err != nil { - return fmt.Errorf("exact workflow run %s source: %w", runID, err) - } - sourceUpdatedAt, err = latestWorkflowTimestamp(sourceUpdatedAt, exactSource) - if err != nil { - return err - } - if workflowTimestampBefore(existingSource, exactSource) { - merged := make(map[string]any, len(existing)+len(exact)) - for key, value := range existing { - merged[key] = value - } - for key, value := range exact { - merged[key] = value + switch { + case workflowTimestampBefore(existingSource, incomingSource): + rowsByID[runID] = row + case workflowTimestampBefore(incomingSource, existingSource): + default: + if mustJSON(existing) != mustJSON(row) { + return fmt.Errorf( + "conflicting workflow run %s observations share one sync generation", + runID, + ) } - rowsByID[runID] = merged } } - for runID := range deletedRunIDs { - delete(rowsByID, runID) + for runID := range observedRunIDs { + if _, present := presentRunIDs[runID]; !present { + laterAbsentRunIDs[runID] = struct{}{} + } } - - runIDs := make([]string, 0, len(rowsByID)) - for runID := range rowsByID { - runIDs = append(runIDs, runID) + for runID := range presentRunIDs { + observedRunIDs[runID] = struct{}{} } - sort.Strings(runIDs) - consolidatedRows := make([]map[string]any, 0, len(runIDs)) - for _, runID := range runIDs { - consolidatedRows = append(consolidatedRows, rowsByID[runID]) + } + for runID := range laterAbsentRunIDs { + if _, deleted := deletedRunIDs[runID]; deleted { + continue } - tombstones := make([]string, 0, len(deletedRunIDs)) - for runID := range deletedRunIDs { - tombstones = append(tombstones, runID) + if !allowRequests { + return errWorkflowVerificationNotAttempted } - sort.Strings(tombstones) - for _, index := range group.indices { - rows := &payloads[index].pullDetails - rows.workflowSourceUpdatedAt = sourceUpdatedAt - rows.workflowDeletedRunIDs = append([]string(nil), tombstones...) - rows.runsRaw = append([]map[string]any(nil), consolidatedRows...) + exact, deleted, err := s.verifySiblingWorkflowRunDeletion( + ctx, + options, + headSHA, + runID, + ) + if err != nil { + return err + } + if deleted { + deletedRunIDs[runID] = struct{}{} + continue } + existing := rowsByID[runID] + existingSource, err := workflowRunTimestamp( + stringValue(existing["updated_at"]), + stringValue(existing["created_at"]), + ) + if err != nil { + return fmt.Errorf("workflow run %s source: %w", runID, err) + } + exactSource, err := workflowRunTimestamp( + stringValue(exact["updated_at"]), + stringValue(exact["created_at"]), + ) + if err != nil { + return fmt.Errorf("exact workflow run %s source: %w", runID, err) + } + sourceUpdatedAt, err = latestWorkflowTimestamp(sourceUpdatedAt, exactSource) + if err != nil { + return err + } + if workflowTimestampBefore(existingSource, exactSource) { + merged := make(map[string]any, len(existing)+len(exact)) + for key, value := range existing { + merged[key] = value + } + for key, value := range exact { + merged[key] = value + } + rowsByID[runID] = merged + } + } + for runID := range deletedRunIDs { + delete(rowsByID, runID) + } + + runIDs := make([]string, 0, len(rowsByID)) + for runID := range rowsByID { + runIDs = append(runIDs, runID) + } + sort.Strings(runIDs) + consolidatedRows := make([]map[string]any, 0, len(runIDs)) + for _, runID := range runIDs { + consolidatedRows = append(consolidatedRows, rowsByID[runID]) + } + tombstones := make([]string, 0, len(deletedRunIDs)) + for runID := range deletedRunIDs { + tombstones = append(tombstones, runID) + } + sort.Strings(tombstones) + for _, index := range indices { + rows := &payloads[index].pullDetails + rows.workflowSourceUpdatedAt = sourceUpdatedAt + rows.workflowDeletedRunIDs = append([]string(nil), tombstones...) + rows.runsRaw = append([]map[string]any(nil), consolidatedRows...) } return nil } diff --git a/internal/syncer/syncer.go b/internal/syncer/syncer.go index c0512d1..c605bda 100644 --- a/internal/syncer/syncer.go +++ b/internal/syncer/syncer.go @@ -98,7 +98,6 @@ type syncPersistStats struct { FingerprintsUpserted int ThreadsClosed int ThreadsSkippedStale int - FinishedAt string } type threadSyncPayload struct { @@ -138,13 +137,42 @@ func (s *Syncer) Sync(ctx context.Context, options Options) (Stats, error) { if err != nil { return Stats{}, err } + if err := s.store.WithTx(ctx, func(st *store.Store) error { + repoID, err := s.upsertRepository(ctx, st, options, repoRaw) + if err != nil { + return err + } + return st.PreserveClosedSweepWatermark(ctx, repoID, startedAt) + }); err != nil { + return Stats{}, err + } + var failures []error + var reserveErr *gh.RateLimitReserveError + recordFailure := func(number int, row map[string]any, cause error, operations ...string) error { + failures = append(failures, fmt.Errorf("%s #%d: %w", strings.Join(operations, ", "), number, cause)) + errors.As(cause, &reserveErr) + if err := s.recordSyncFailure(ctx, options, repoRaw, row, number, cause, operations...); err != nil { + failures = append(failures, fmt.Errorf("record sync attempt failure: %w", err)) + return errors.Join(failures...) + } + if ctx.Err() != nil { + return errors.Join(failures...) + } + return nil + } numbers := uniquePositiveNumbers(options.Numbers) rows := make([]map[string]any, 0, len(numbers)) if len(numbers) > 0 { for _, number := range numbers { + if reserveErr != nil { + break + } row, err := s.client.GetIssue(ctx, options.Owner, options.Repo, number, options.Reporter) if err != nil { - return Stats{}, err + if err := recordFailure(number, nil, err, "issue"); err != nil { + return Stats{}, err + } + continue } rows = append(rows, row) } @@ -206,10 +234,17 @@ func (s *Syncer) Sync(ctx context.Context, options Options) (Stats, error) { payload := threadSyncPayload{row: row} number := intValue(row["number"]) kind := issueKind(row) + if reserveErr != nil && (options.IncludeComments || + kind == "pull_request" && (options.IncludePRMetadata || options.IncludePRDetails)) { + continue + } if options.IncludeComments { - commentRows, err := s.fetchCommentRows(ctx, options, kind, number) + commentRows, operation, err := s.fetchCommentRows(ctx, options, kind, number) if err != nil { - return Stats{}, err + if err := recordFailure(number, row, err, operation); err != nil { + return Stats{}, err + } + continue } payload.commentRows = commentRows received.CommentsReceived += len(commentRows) @@ -217,19 +252,19 @@ func (s *Syncer) Sync(ctx context.Context, options Options) (Stats, error) { if options.IncludePRDetails && kind == "pull_request" { reviewThreads, reviewThreadsFetchedAt, err := s.fetchPullReviewThreadRows(ctx, options, number) if err != nil { - if recordErr := s.recordPullRequestSyncFailure(ctx, options, repoRaw, row, "pull_review_threads", err); recordErr != nil { - return Stats{}, fmt.Errorf("%w; additionally failed to record sync attempt failure: %v", err, recordErr) + if err := recordFailure(number, row, err, "pull_review_threads"); err != nil { + return Stats{}, err } - return Stats{}, err + continue } payload.reviewThreads = reviewThreads payload.reviewThreadsFetchedAt = reviewThreadsFetchedAt pullDetails, err := s.fetchPullRequestDetails(ctx, options, number) if err != nil { - if recordErr := s.recordPullRequestSyncFailure(ctx, options, repoRaw, row, "pull_request_details", err); recordErr != nil { - return Stats{}, fmt.Errorf("%w; additionally failed to record sync attempt failure: %v", err, recordErr) + if err := recordFailure(number, row, err, "pull_request_details"); err != nil { + return Stats{}, err } - return Stats{}, err + continue } workflowObservationOrder++ pullDetails.workflowObservationOrder = workflowObservationOrder @@ -238,10 +273,10 @@ func (s *Syncer) Sync(ctx context.Context, options Options) (Stats, error) { } else if options.IncludePRMetadata && kind == "pull_request" { pullDetails, err := s.fetchPullRequestMetadata(ctx, options, number) if err != nil { - if recordErr := s.recordPullRequestSyncFailure(ctx, options, repoRaw, row, "pull_request_metadata", err); recordErr != nil { - return Stats{}, fmt.Errorf("%w; additionally failed to record sync attempt failure: %v", err, recordErr) + if err := recordFailure(number, row, err, "pull_request_metadata"); err != nil { + return Stats{}, err } - return Stats{}, err + continue } payload.pullDetails = pullDetails payload.hasPullDetails = true @@ -251,9 +286,20 @@ func (s *Syncer) Sync(ctx context.Context, options Options) (Stats, error) { return Stats{}, err } } - if err := s.consolidateWorkflowSnapshots(ctx, options, payloads); err != nil { - return Stats{}, err + groupFailures := s.consolidateWorkflowSnapshots(ctx, options, payloads, reserveErr == nil) + completed := payloads[:0] + for index, payload := range payloads { + if cause, excluded := groupFailures[index]; excluded { + if cause != nil { + if err := recordFailure(intValue(payload.row["number"]), payload.row, cause, "pull_request_details"); err != nil { + return Stats{}, err + } + } + continue + } + completed = append(completed, payload) } + payloads = completed // Order replace-all child snapshots when their complete observations are available. observationSequence, err := s.store.NextThreadObservationSequence(ctx, started) if err != nil { @@ -280,9 +326,6 @@ func (s *Syncer) Sync(ctx context.Context, options Options) (Stats, error) { MetadataOnly: !options.IncludeComments && !options.IncludePRMetadata && !options.IncludePRDetails, StartedAt: started, } - if len(numbers) == 0 && options.Limit <= 0 && since == "" { - stats.ClosedSweepThrough = started - } tracker := progress.New(options.Logger, progress.Options{ Name: "sync", Unit: "threads", @@ -292,21 +335,17 @@ func (s *Syncer) Sync(ctx context.Context, options Options) (Stats, error) { "state", state, }, }) - var persisted syncPersistStats - persist := func(st *store.Store) error { - attempt := syncPersistStats{} - repoID, err := st.UpsertRepository(ctx, store.Repository{ - Owner: options.Owner, - Name: options.Repo, - FullName: options.Owner + "/" + options.Repo, - GitHubRepoID: jsonID(repoRaw["id"]), - RawJSON: mustJSON(repoRaw), - UpdatedAt: s.now().Format(time.RFC3339Nano), - }) - if err != nil { - return err - } - for _, payload := range payloads { + // Each complete item owns its transaction, but shared-head observations above + // retain one consolidated snapshot and generation across those transactions. + for _, payload := range payloads { + var attempt syncPersistStats + err := s.store.WithTx(ctx, func(st *store.Store) error { + // WithTx can retry a busy transaction; only its committed attempt counts. + attempt = syncPersistStats{} + repoID, err := s.upsertRepository(ctx, st, options, repoRaw) + if err != nil { + return err + } thread := mapIssueToThread(repoID, payload.row, s.now().Format(time.RFC3339Nano)) _, hasIssueDraft := payload.row["draft"] if payload.hasPullDetails { @@ -344,13 +383,7 @@ func (s *Syncer) Sync(ctx context.Context, options Options) (Stats, error) { attempt.WorkflowRunsSynced += count } attempt.ThreadsSkippedStale++ - tracker.Add(1, - "number", thread.Number, - "kind", thread.Kind, - "thread_state", thread.State, - "result", "stale_skipped", - ) - continue + return nil } thread.ID = upsert.ID completeEvidence := hasFreshThreadEvidence(options, thread) @@ -379,6 +412,16 @@ func (s *Syncer) Sync(ctx context.Context, options Options) (Stats, error) { return err } } + var resolvedOperations []string + if upsert.Applied { + resolvedOperations = append(resolvedOperations, "issue") + } + if childReservations[store.ThreadChildComments] { + resolvedOperations = append(resolvedOperations, "issue_comments") + if thread.Kind == "pull_request" { + resolvedOperations = append(resolvedOperations, "pull_reviews", "pull_review_comments") + } + } if payload.hasPullDetails { if err := reserveChild(store.ThreadChildPullRequestDetails); err != nil { return err @@ -431,6 +474,7 @@ func (s *Syncer) Sync(ctx context.Context, options Options) (Stats, error) { return err } attempt.ReviewThreadsSynced += count + resolvedOperations = append(resolvedOperations, "pull_review_threads") } detailStats, err := s.persistPullRequestDetails( ctx, @@ -450,16 +494,14 @@ func (s *Syncer) Sync(ctx context.Context, options Options) (Stats, error) { } if detailStats.details { attempt.PRDetailsSynced++ + resolvedOperations = append(resolvedOperations, "pull_request_metadata") } - // Metadata cannot repair failures from child collections it did not fetch. - var operations []string - if !options.IncludePRDetails { - operations = []string{"pull_request_metadata"} - } - if options.IncludePRDetails || detailStats.details { - if _, err := st.ResolveSyncAttemptFailures(ctx, repoID, thread.Number, s.now().Format(time.RFC3339Nano), operations...); err != nil { - return err - } + if detailStats.details && + childReservations[store.ThreadChildPullRequestFiles] && + childReservations[store.ThreadChildPullRequestCommits] && + childReservations[store.ThreadChildPullRequestChecks] && + payload.pullDetails.workflowSnapshotFresh { + resolvedOperations = append(resolvedOperations, "pull_request_details") } attempt.PRFilesSynced += detailStats.files attempt.PRCommitsSynced += detailStats.commits @@ -509,40 +551,66 @@ func (s *Syncer) Sync(ctx context.Context, options Options) (Stats, error) { } else { attempt.IssuesSynced++ } - tracker.Add(1, - "number", thread.Number, - "kind", thread.Kind, - "thread_state", thread.State, - ) + // Resolve only observed families, atomically with their persisted data. + // An empty operation filter would resolve every family in the store API. + if len(resolvedOperations) > 0 { + _, err = st.ResolveSyncAttemptFailures(ctx, repoID, thread.Number, s.now().Format(time.RFC3339Nano), resolvedOperations...) + } + return err + }) + if err != nil { + // The failed transaction is gone. Record only its existing parent, + // under each requested family so a shallow retry cannot clear children. + operations := requestedSyncOperations(options, issueKind(payload.row)) + if recordErr := recordFailure(intValue(payload.row["number"]), nil, err, operations...); recordErr != nil { + break + } + continue } - if attempt.ThreadsClosed > 0 { - options.Reporter.Printf( - "[sync] closed overlap sweep matched %d stale open thread(s)", - attempt.ThreadsClosed, - ) + addSyncPersistStats(&stats, attempt) + tracker.Add(1, + "number", intValue(payload.row["number"]), + "kind", issueKind(payload.row), + "thread_state", stringValue(payload.row["state"]), + "stale_skipped", attempt.ThreadsSkippedStale > 0, + ) + } + stats.FinishedAt = s.now().Format(time.RFC3339Nano) + if stats.ThreadsClosed > 0 { + options.Reporter.Printf( + "[sync] closed overlap sweep matched %d stale open thread(s)", + stats.ThreadsClosed, + ) + } + if err := errors.Join(failures...); err != nil { + tracker.Finish(err) + return stats, err + } + if len(numbers) == 0 && options.Limit <= 0 && since == "" { + stats.ClosedSweepThrough = started + } + // Partial commits are useful evidence, never a successful list checkpoint. + err = s.store.WithTx(ctx, func(st *store.Store) error { + repoID, err := s.upsertRepository(ctx, st, options, repoRaw) + if err != nil { + return err } - attempt.FinishedAt = s.now().Format(time.RFC3339Nano) - runStats := stats - applySyncPersistStats(&runStats, attempt) - if _, err := st.RecordRun(ctx, store.RunRecord{ + _, err = st.RecordRun(ctx, store.RunRecord{ RepoID: repoID, Kind: "sync", Scope: syncRunScope(state, numbers), Status: "success", - StartedAt: runStats.StartedAt, - FinishedAt: runStats.FinishedAt, - StatsJSON: mustJSON(runStats), - }); err != nil { - return err - } - persisted = attempt - return nil - } - if err := s.store.WithTx(ctx, persist); err != nil { + StartedAt: stats.StartedAt, + FinishedAt: stats.FinishedAt, + StatsJSON: mustJSON(stats), + }) + return err + }) + if err != nil { + stats.ClosedSweepThrough = "" tracker.Finish(err) - return Stats{}, err + return stats, err } - applySyncPersistStats(&stats, persisted) tracker.Finish(nil) return stats, nil } @@ -569,7 +637,18 @@ func reportSyncProgress( return reporter(snapshot) } -func (s *Syncer) recordPullRequestSyncFailure(ctx context.Context, options Options, repoRaw, row map[string]any, operation string, syncErr error) error { +func (s *Syncer) upsertRepository(ctx context.Context, st *store.Store, options Options, repoRaw map[string]any) (int64, error) { + return st.UpsertRepository(ctx, store.Repository{ + Owner: options.Owner, + Name: options.Repo, + FullName: options.Owner + "/" + options.Repo, + GitHubRepoID: jsonID(repoRaw["id"]), + RawJSON: mustJSON(repoRaw), + UpdatedAt: s.now().Format(time.RFC3339Nano), + }) +} + +func (s *Syncer) recordSyncFailure(ctx context.Context, options Options, repoRaw, row map[string]any, number int, syncErr error, operations ...string) error { if syncErr == nil { return nil } @@ -577,33 +656,55 @@ func (s *Syncer) recordPullRequestSyncFailure(ctx context.Context, options Optio defer cancel() return s.store.WithTx(recordCtx, func(st *store.Store) error { now := s.now().Format(time.RFC3339Nano) - repoID, err := st.UpsertRepository(recordCtx, store.Repository{ - Owner: options.Owner, - Name: options.Repo, - FullName: options.Owner + "/" + options.Repo, - GitHubRepoID: jsonID(repoRaw["id"]), - RawJSON: mustJSON(repoRaw), - UpdatedAt: now, - }) + repoID, err := s.upsertRepository(recordCtx, st, options, repoRaw) if err != nil { return err } - thread := mapIssueToThread(repoID, row, now) - threadID, err := st.UpsertThread(recordCtx, thread) - if err != nil { - return err + var threadID int64 + if row != nil { + thread := mapIssueToThread(repoID, row, now) + _, hasDraft := row["draft"] + upsert, err := st.UpsertThreadObservation(recordCtx, thread, store.UpsertThreadOptions{ + IncompleteEvidence: true, + PreserveDraft: thread.Kind == "pull_request" && !hasDraft, + }) + if err != nil { + return err + } + threadID = upsert.ID + // The parent is retained even when children fail. Resolve only its + // applied observation, atomically with the failed-child bookkeeping. + if upsert.Applied { + if _, err := st.ResolveSyncAttemptFailures(recordCtx, repoID, number, now, "issue"); err != nil { + return err + } + } + } else { + threads, err := st.ListThreadsFiltered(recordCtx, store.ThreadListOptions{ + RepoID: repoID, IncludeClosed: true, Numbers: []int{number}, Limit: 1, + }) + if err != nil { + return err + } + if len(threads) > 0 { + threadID = threads[0].ID + } } - _, err = st.RecordSyncAttemptFailure(recordCtx, store.SyncAttemptFailure{ - RepoID: repoID, - ThreadID: threadID, - Number: thread.Number, - Operation: operation, - ErrorClass: syncAttemptErrorClass(syncErr), - ErrorMessage: syncErr.Error(), - FirstSeenAt: now, - LastSeenAt: now, - }) - return err + for _, operation := range operations { + if _, err := st.RecordSyncAttemptFailure(recordCtx, store.SyncAttemptFailure{ + RepoID: repoID, + ThreadID: threadID, + Number: number, + Operation: operation, + ErrorClass: syncAttemptErrorClass(syncErr), + ErrorMessage: syncErr.Error(), + FirstSeenAt: now, + LastSeenAt: now, + }); err != nil { + return err + } + } + return nil }) } @@ -618,23 +719,22 @@ func syncAttemptErrorClass(err error) string { } } -func applySyncPersistStats(stats *Stats, persisted syncPersistStats) { - stats.ThreadsSynced = persisted.ThreadsSynced - stats.IssuesSynced = persisted.IssuesSynced - stats.PullRequestsSynced = persisted.PullRequestsSynced - stats.CommentsSynced = persisted.CommentsSynced - stats.ReviewThreadsSynced = persisted.ReviewThreadsSynced - stats.PRDetailsSynced = persisted.PRDetailsSynced - stats.PRFilesSynced = persisted.PRFilesSynced - stats.PRCommitsSynced = persisted.PRCommitsSynced - stats.PRChecksSynced = persisted.PRChecksSynced - stats.WorkflowRunsSynced = persisted.WorkflowRunsSynced - stats.EvidenceObserved = persisted.EvidenceObserved - stats.RevisionsCreated = persisted.RevisionsCreated - stats.FingerprintsUpserted = persisted.FingerprintsUpserted - stats.ThreadsClosed = persisted.ThreadsClosed - stats.ThreadsSkippedStale = persisted.ThreadsSkippedStale - stats.FinishedAt = persisted.FinishedAt +func addSyncPersistStats(stats *Stats, persisted syncPersistStats) { + stats.ThreadsSynced += persisted.ThreadsSynced + stats.IssuesSynced += persisted.IssuesSynced + stats.PullRequestsSynced += persisted.PullRequestsSynced + stats.CommentsSynced += persisted.CommentsSynced + stats.ReviewThreadsSynced += persisted.ReviewThreadsSynced + stats.PRDetailsSynced += persisted.PRDetailsSynced + stats.PRFilesSynced += persisted.PRFilesSynced + stats.PRCommitsSynced += persisted.PRCommitsSynced + stats.PRChecksSynced += persisted.PRChecksSynced + stats.WorkflowRunsSynced += persisted.WorkflowRunsSynced + stats.EvidenceObserved += persisted.EvidenceObserved + stats.RevisionsCreated += persisted.RevisionsCreated + stats.FingerprintsUpserted += persisted.FingerprintsUpserted + stats.ThreadsClosed += persisted.ThreadsClosed + stats.ThreadsSkippedStale += persisted.ThreadsSkippedStale } func uniquePositiveNumbers(numbers []int) []int { @@ -719,6 +819,25 @@ func hasFreshThreadEvidence(options Options, thread store.Thread) bool { return options.IncludeComments && (thread.Kind != "pull_request" || options.IncludePRDetails) } +func requestedSyncOperations(options Options, kind string) []string { + operations := []string{"issue"} + if options.IncludeComments { + operations = append(operations, "issue_comments") + if kind == "pull_request" { + operations = append(operations, "pull_reviews", "pull_review_comments") + } + } + if kind == "pull_request" { + if options.IncludePRMetadata || options.IncludePRDetails { + operations = append(operations, "pull_request_metadata") + } + if options.IncludePRDetails { + operations = append(operations, "pull_review_threads", "pull_request_details") + } + } + return operations +} + func normalizeSince(value string, now time.Time) (string, error) { value = strings.TrimSpace(value) if value == "" { @@ -860,11 +979,11 @@ func persistThreadEnrichment( return st.UpsertThreadRevisionAndFingerprint(ctx, evidence, createdAt) } -func (s *Syncer) fetchCommentRows(ctx context.Context, options Options, threadKind string, number int) ([]commentRow, error) { +func (s *Syncer) fetchCommentRows(ctx context.Context, options Options, threadKind string, number int) ([]commentRow, string, error) { var rows []commentRow issueComments, err := s.client.ListIssueComments(ctx, options.Owner, options.Repo, number, options.Reporter) if err != nil { - return nil, err + return nil, "issue_comments", err } for _, row := range issueComments { rows = append(rows, commentRow{kind: "issue_comment", raw: row}) @@ -872,20 +991,20 @@ func (s *Syncer) fetchCommentRows(ctx context.Context, options Options, threadKi if threadKind == "pull_request" { reviews, err := s.client.ListPullReviews(ctx, options.Owner, options.Repo, number, options.Reporter) if err != nil { - return nil, err + return nil, "pull_reviews", err } for _, row := range reviews { rows = append(rows, commentRow{kind: "pull_review", raw: row}) } reviewComments, err := s.client.ListPullReviewComments(ctx, options.Owner, options.Repo, number, options.Reporter) if err != nil { - return nil, err + return nil, "pull_review_comments", err } for _, row := range reviewComments { rows = append(rows, commentRow{kind: "pull_review_comment", raw: row}) } } - return rows, nil + return rows, "", nil } func persistComments( diff --git a/internal/syncer/syncer_test.go b/internal/syncer/syncer_test.go index dc26bec..9573d29 100644 --- a/internal/syncer/syncer_test.go +++ b/internal/syncer/syncer_test.go @@ -3,6 +3,7 @@ package syncer import ( "bytes" "context" + "database/sql" "encoding/json" "errors" "fmt" @@ -1245,9 +1246,15 @@ func (failingSecondCommentGitHub) ListIssueComments(ctx context.Context, owner, } func (g *txProbePullDetailsGitHub) ListPullFiles(ctx context.Context, owner, repo string, number int, reporter gh.Reporter) ([]map[string]any, error) { + ctx, cancel := context.WithTimeout(ctx, time.Second) + defer cancel() storedRepo, err := g.st.RepositoryByFullName(ctx, owner+"/"+repo) if err != nil { - g.sawMissingPersistedThread = true + if errors.Is(err, sql.ErrNoRows) { + g.sawMissingPersistedThread = true + } else { + g.sawPersistedThreadReadErr = err + } return nil, nil } threads, err := g.st.ListThreads(ctx, storedRepo.ID, true) @@ -1260,6 +1267,7 @@ func (g *txProbePullDetailsGitHub) ListPullFiles(ctx context.Context, owner, rep g.sawPersistedThread = true } } + g.sawMissingPersistedThread = !g.sawPersistedThread return nil, nil } @@ -2687,7 +2695,7 @@ func TestConsolidateWorkflowSnapshotsTombstonesDeletionAcrossStaleReappearanceOr if err := s.consolidateWorkflowSnapshots(ctx, Options{ Owner: "openclaw", Repo: "gitcrawl", - }, payloads); err != nil { + }, payloads, true); err != nil { t.Fatalf("consolidate workflow snapshots: %v", err) } for index := range payloads { @@ -3183,7 +3191,7 @@ func TestSyncRollsBackThreadRevisionWhenFingerprintFails(t *testing.T) { if _, err := s.Sync(ctx, Options{Owner: "openclaw", Repo: "gitcrawl", IncludeComments: true, IncludePRDetails: true}); err == nil || !strings.Contains(err.Error(), "fingerprint rejected") { t.Fatalf("sync error = %v", err) } - for _, table := range []string{"repositories", "threads", "thread_revisions", "thread_fingerprints"} { + for _, table := range []string{"threads", "thread_revisions", "thread_fingerprints"} { var count int if err := st.DB().QueryRowContext(ctx, `select count(*) from `+table).Scan(&count); err != nil { t.Fatalf("%s count: %v", table, err) @@ -3194,7 +3202,7 @@ func TestSyncRollsBackThreadRevisionWhenFingerprintFails(t *testing.T) { } } -func TestSyncWithCommentsRollsBackOnCommentFetchError(t *testing.T) { +func TestSyncWithCommentsRetainsCompletedItemsOnCommentFetchError(t *testing.T) { ctx := context.Background() st, err := store.Open(ctx, filepath.Join(t.TempDir(), "gitcrawl.db")) if err != nil { @@ -3204,18 +3212,19 @@ func TestSyncWithCommentsRollsBackOnCommentFetchError(t *testing.T) { s := New(failingSecondCommentGitHub{}, st) s.now = func() time.Time { return time.Date(2026, 4, 26, 0, 0, 0, 0, time.UTC) } - _, err = s.Sync(ctx, Options{Owner: "openclaw", Repo: "gitcrawl", IncludeComments: true}) + stats, err := s.Sync(ctx, Options{Owner: "openclaw", Repo: "gitcrawl", IncludeComments: true}) if err == nil || !strings.Contains(err.Error(), "comments unavailable") { t.Fatalf("sync error = %v", err) } - if _, err := st.RepositoryByFullName(ctx, "openclaw/gitcrawl"); err == nil { - t.Fatal("repository persisted after failed comment hydration") + if stats.ThreadsSynced != 1 || stats.CommentsSynced != 1 || stats.ClosedSweepThrough != "" { + t.Fatalf("committed stats = %+v", stats) } - assertTableRowCount(t, st, "repositories", 0) - assertTableRowCount(t, st, "threads", 0) - assertTableRowCount(t, st, "comments", 0) - assertTableRowCount(t, st, "documents", 0) - assertTableRowCount(t, st, "sync_runs", 0) + assertTableRowCount(t, st, "repositories", 1) + assertTableRowCount(t, st, "threads", 2) + assertTableRowCount(t, st, "comments", 1) + assertTableRowCount(t, st, "documents", 1) + assertTableRowCount(t, st, "sync_attempt_failures", 1) + assertTableRowCount(t, st, "sync_runs", 1) } func TestMetadataOnlySyncPreservesCommentBackedDocumentText(t *testing.T) { @@ -3715,7 +3724,7 @@ func TestSyncPullRequestDetailsFailsOnReviewThreadFetchError(t *testing.T) { t.Fatalf("failure = %+v", failures[0]) } assertTableRowCount(t, st, "pull_request_review_thread_syncs", 0) - assertTableRowCount(t, st, "sync_runs", 0) + assertTableRowCount(t, st, "sync_runs", 1) s = New(pullDetailsGitHub{}, st) s.now = func() time.Time { return time.Date(2026, 4, 26, 0, 1, 0, 0, time.UTC) } @@ -3763,7 +3772,7 @@ func TestSyncAttemptErrorClass(t *testing.T) { } } -func TestRecordPullRequestSyncFailureOutlivesCanceledFetchContext(t *testing.T) { +func TestRecordSyncFailureOutlivesCanceledFetchContext(t *testing.T) { background := context.Background() st, err := store.Open(background, filepath.Join(t.TempDir(), "gitcrawl.db")) if err != nil { @@ -3782,7 +3791,7 @@ func TestRecordPullRequestSyncFailureOutlivesCanceledFetchContext(t *testing.T) } canceled, cancel := context.WithCancel(background) cancel() - if err := s.recordPullRequestSyncFailure(canceled, Options{Owner: "openclaw", Repo: "gitcrawl"}, repoRaw, row, "pull_request_details", context.Canceled); err != nil { + if err := s.recordSyncFailure(canceled, Options{Owner: "openclaw", Repo: "gitcrawl"}, repoRaw, row, 8, context.Canceled, "pull_request_details"); err != nil { t.Fatalf("record canceled sync failure: %v", err) } repo, err := st.RepositoryByFullName(background, "openclaw/gitcrawl") @@ -3840,7 +3849,7 @@ func TestSyncPullRequestDetailsDoesNotFetchInsideTransaction(t *testing.T) { t.Fatalf("probe persisted thread: %v", client.sawPersistedThreadReadErr) } if !client.sawMissingPersistedThread { - t.Fatal("PR detail fetch saw repository writes before hydration finished") + t.Fatal("PR detail fetch saw thread writes before hydration finished") } if client.sawPersistedThread { t.Fatal("PR detail fetch saw persisted PR thread before hydration finished")