From 169ef5f63b6f5594fe7b45a01cf3bc64410ac325 Mon Sep 17 00:00:00 2001 From: Andy Ye <35905412+TurboTheTurtle@users.noreply.github.com> Date: Wed, 1 Jul 2026 12:29:54 -0700 Subject: [PATCH 1/3] Track sync hydration failures --- internal/cli/app.go | 55 +++++++++ internal/cli/app_test.go | 79 +++++++++++++ internal/store/schema.go | 17 +++ internal/store/sqlc/schema.sql | 15 +++ internal/store/store.go | 2 +- internal/store/sync_failures.go | 160 +++++++++++++++++++++++++++ internal/store/sync_failures_test.go | 90 +++++++++++++++ internal/syncer/syncer.go | 57 ++++++++++ internal/syncer/syncer_test.go | 37 ++++++- 9 files changed, 507 insertions(+), 5 deletions(-) create mode 100644 internal/store/sync_failures.go create mode 100644 internal/store/sync_failures_test.go diff --git a/internal/cli/app.go b/internal/cli/app.go index aad709f..063bd3d 100644 --- a/internal/cli/app.go +++ b/internal/cli/app.go @@ -175,6 +175,8 @@ func (a *App) Run(ctx context.Context, args []string) error { return a.runSetClusterCanonical(ctx, rest[1:]) case "runs": return a.runRuns(ctx, rest[1:]) + case "sync-failures": + return a.runSyncFailures(ctx, rest[1:]) case "search": return a.runSearch(ctx, rest[1:]) case "code": @@ -2256,6 +2258,53 @@ func (a *App) runRuns(ctx context.Context, args []string) error { }, true) } +func (a *App) runSyncFailures(ctx context.Context, args []string) error { + fs := flag.NewFlagSet("sync-failures", flag.ContinueOnError) + fs.SetOutput(io.Discard) + includeResolved := fs.Bool("include-resolved", false, "include resolved failure rows") + limitRaw := fs.String("limit", "", "maximum failure rows") + jsonOut := fs.Bool("json", false, "write JSON output") + if err := fs.Parse(normalizeCommandArgs(args, map[string]bool{"limit": true})); err != nil { + return usageErr(err) + } + a.applyCommandJSON(*jsonOut) + if fs.NArg() != 1 { + return usageErr(fmt.Errorf("sync-failures requires owner/repo")) + } + owner, repoName, err := parseOwnerRepo(fs.Arg(0)) + if err != nil { + return usageErr(err) + } + limit, err := parseOptionalPositiveInt(*limitRaw) + if err != nil { + return usageErr(err) + } + + rt, err := a.openLocalRuntimeReadOnly(ctx) + if err != nil { + return err + } + defer rt.Store.Close() + + repo, err := rt.repository(ctx, owner, repoName) + if err != nil { + return err + } + failures, err := rt.Store.ListSyncAttemptFailures(ctx, store.SyncAttemptFailureListOptions{ + RepoID: repo.ID, + IncludeResolved: *includeResolved, + Limit: limit, + }) + if err != nil { + return err + } + return a.writeOutput("sync-failures", map[string]any{ + "repository": repo.FullName, + "include_resolved": *includeResolved, + "failures": failures, + }, true) +} + func (a *App) runThreads(ctx context.Context, args []string) error { fs := flag.NewFlagSet("threads", flag.ContinueOnError) fs.SetOutput(io.Discard) @@ -4323,6 +4372,7 @@ Core commands: init create config, optionally from a portable store doctor check config, token, and database readiness sync sync GitHub issue and pull request metadata + sync-failures list failed sync hydration attempts refresh run sync, enrichment, embedding, and clustering pipeline embed generate OpenAI embeddings for local thread documents threads list local issue and pull request rows @@ -4406,6 +4456,11 @@ Usage: Usage: gitcrawl sync owner/repo [--state open|closed|all] [--numbers refs] [--with pr-details] [--include-pr-details] [--json] +`, + "sync-failures": `gitcrawl sync-failures lists failed sync hydration attempts. + +Usage: + gitcrawl sync-failures owner/repo [--include-resolved] [--limit N] [--json] `, "refresh": `gitcrawl refresh runs sync, enrichment, embedding, and clustering. diff --git a/internal/cli/app_test.go b/internal/cli/app_test.go index e901d5a..ae98eb8 100644 --- a/internal/cli/app_test.go +++ b/internal/cli/app_test.go @@ -2484,6 +2484,85 @@ func TestTUIInfersRepository(t *testing.T) { } } +func TestSyncFailuresListsUnresolvedAndHistory(t *testing.T) { + ctx := context.Background() + dir := t.TempDir() + configPath := filepath.Join(dir, "config.toml") + dbPath := filepath.Join(dir, "gitcrawl.db") + app := New() + if err := app.Run(ctx, []string{"--config", configPath, "init", "--db", dbPath}); err != nil { + t.Fatalf("init: %v", err) + } + st, err := store.Open(ctx, dbPath) + if err != nil { + t.Fatalf("open store: %v", err) + } + repoID, err := st.UpsertRepository(ctx, store.Repository{ + Owner: "openclaw", + Name: "gitcrawl", + FullName: "openclaw/gitcrawl", + RawJSON: "{}", + UpdatedAt: "2026-06-06T00:00:00Z", + }) + if err != nil { + t.Fatalf("seed repository: %v", err) + } + for _, number := range []int{83, 84} { + threadID, err := st.UpsertThread(ctx, store.Thread{ + RepoID: repoID, GitHubID: strconv.Itoa(number), Number: number, Kind: "pull_request", State: "open", + Title: "hydration failure", HTMLURL: fmt.Sprintf("https://github.com/openclaw/gitcrawl/pull/%d", number), + LabelsJSON: "[]", AssigneesJSON: "[]", RawJSON: "{}", ContentHash: fmt.Sprintf("h%d", number), UpdatedAt: "2026-06-06T00:00:00Z", + }) + if err != nil { + t.Fatalf("seed thread %d: %v", number, err) + } + if _, err := st.RecordSyncAttemptFailure(ctx, store.SyncAttemptFailure{ + RepoID: repoID, ThreadID: threadID, Number: number, Operation: "pull_request_details", ErrorClass: "error", + ErrorMessage: fmt.Sprintf("failure %d", number), FirstSeenAt: "2026-06-06T00:00:00Z", LastSeenAt: "2026-06-06T00:00:00Z", + }); err != nil { + t.Fatalf("seed failure %d: %v", number, err) + } + if number == 83 { + if _, err := st.ResolveSyncAttemptFailures(ctx, repoID, number, "2026-06-06T00:05:00Z"); err != nil { + t.Fatalf("resolve failure %d: %v", number, err) + } + } + } + if err := st.Close(); err != nil { + t.Fatalf("close store: %v", err) + } + + run := New() + var stdout bytes.Buffer + run.Stdout = &stdout + if err := run.Run(ctx, []string{"--config", configPath, "--json", "sync-failures", "openclaw/gitcrawl"}); err != nil { + t.Fatalf("sync-failures: %v", err) + } + var active struct { + Failures []store.SyncAttemptFailure `json:"failures"` + } + if err := json.Unmarshal(stdout.Bytes(), &active); err != nil { + t.Fatalf("decode active failures: %v\n%s", err, stdout.String()) + } + if len(active.Failures) != 1 || active.Failures[0].Number != 84 || active.Failures[0].ResolvedAt != "" { + t.Fatalf("active failures = %+v", active.Failures) + } + + stdout.Reset() + if err := run.Run(ctx, []string{"--config", configPath, "--json", "sync-failures", "openclaw/gitcrawl", "--include-resolved"}); err != nil { + t.Fatalf("sync-failures history: %v", err) + } + var history struct { + Failures []store.SyncAttemptFailure `json:"failures"` + } + if err := json.Unmarshal(stdout.Bytes(), &history); err != nil { + t.Fatalf("decode failure history: %v\n%s", err, stdout.String()) + } + if len(history.Failures) != 2 { + t.Fatalf("history failures = %+v", history.Failures) + } +} + func TestTUIJSONUsesDefaultsWhenConfigMissing(t *testing.T) { ctx := context.Background() dir := t.TempDir() diff --git a/internal/store/schema.go b/internal/store/schema.go index a131878..729fb25 100644 --- a/internal/store/schema.go +++ b/internal/store/schema.go @@ -401,6 +401,21 @@ create table if not exists sync_runs ( error_text text ); +create table if not exists sync_attempt_failures ( + id integer primary key, + repo_id integer not null references repositories(id) on delete cascade, + thread_id integer references threads(id) on delete set null, + number integer not null, + operation text not null, + error_class text not null, + error_message text not null, + first_seen_at text not null, + last_seen_at text not null, + retry_count integer not null default 0, + resolved_at text, + unique(repo_id, number, operation, error_class) +); + create table if not exists summary_runs ( id integer primary key, repo_id integer references repositories(id) on delete cascade, @@ -566,6 +581,8 @@ create index if not exists idx_code_documents_repo_path on code_documents(repo_i create index if not exists idx_thread_fingerprints_hash on thread_fingerprints(fingerprint_hash); create index if not exists idx_thread_vectors_basis_model on thread_vectors(basis, model); create index if not exists idx_sync_runs_repo_status_id on sync_runs(repo_id, status, id); +create index if not exists idx_sync_attempt_failures_repo_unresolved on sync_attempt_failures(repo_id, resolved_at, last_seen_at); +create index if not exists idx_sync_attempt_failures_thread on sync_attempt_failures(thread_id, resolved_at); create index if not exists idx_cluster_runs_repo_status_id on cluster_runs(repo_id, status, id); create index if not exists idx_similarity_edges_repo_score on similarity_edges(repo_id, score); create index if not exists idx_cluster_groups_repo_status on cluster_groups(repo_id, status); diff --git a/internal/store/sqlc/schema.sql b/internal/store/sqlc/schema.sql index 17240eb..897cf5b 100644 --- a/internal/store/sqlc/schema.sql +++ b/internal/store/sqlc/schema.sql @@ -262,6 +262,21 @@ create table sync_runs ( error_text text ); +create table sync_attempt_failures ( + id integer primary key, + repo_id integer not null references repositories(id) on delete cascade, + thread_id integer references threads(id) on delete set null, + number integer not null, + operation text not null, + error_class text not null, + error_message text not null, + first_seen_at text not null, + last_seen_at text not null, + retry_count integer not null default 0, + resolved_at text, + unique(repo_id, number, operation, error_class) +); + create table summary_runs ( id integer primary key, repo_id integer not null references repositories(id) on delete cascade, diff --git a/internal/store/store.go b/internal/store/store.go index 2df20b2..55b6766 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -13,7 +13,7 @@ import ( ) const ( - schemaVersion = 4 + schemaVersion = 5 timeLayout = time.RFC3339Nano ) diff --git a/internal/store/sync_failures.go b/internal/store/sync_failures.go new file mode 100644 index 0000000..8677018 --- /dev/null +++ b/internal/store/sync_failures.go @@ -0,0 +1,160 @@ +package store + +import ( + "context" + "database/sql" + "fmt" + "strings" +) + +type SyncAttemptFailure struct { + ID int64 `json:"id"` + RepoID int64 `json:"repo_id"` + ThreadID int64 `json:"thread_id,omitempty"` + Number int `json:"number"` + Operation string `json:"operation"` + ErrorClass string `json:"error_class"` + ErrorMessage string `json:"error_message"` + FirstSeenAt string `json:"first_seen_at"` + LastSeenAt string `json:"last_seen_at"` + RetryCount int `json:"retry_count"` + ResolvedAt string `json:"resolved_at,omitempty"` +} + +type SyncAttemptFailureListOptions struct { + RepoID int64 + IncludeResolved bool + Limit int +} + +func (s *Store) RecordSyncAttemptFailure(ctx context.Context, failure SyncAttemptFailure) (int64, error) { + if failure.RepoID == 0 { + return 0, fmt.Errorf("record sync attempt failure: missing repo id") + } + if failure.Number <= 0 { + return 0, fmt.Errorf("record sync attempt failure: missing issue or pull request number") + } + operation := strings.TrimSpace(failure.Operation) + if operation == "" { + return 0, fmt.Errorf("record sync attempt failure: missing operation") + } + errorClass := strings.TrimSpace(failure.ErrorClass) + if errorClass == "" { + errorClass = "error" + } + message := strings.TrimSpace(failure.ErrorMessage) + if message == "" { + message = errorClass + } + firstSeen := strings.TrimSpace(failure.FirstSeenAt) + if firstSeen == "" { + firstSeen = failure.LastSeenAt + } + lastSeen := strings.TrimSpace(failure.LastSeenAt) + if lastSeen == "" { + lastSeen = firstSeen + } + var threadID any + if failure.ThreadID != 0 { + threadID = failure.ThreadID + } + if _, err := s.q().ExecContext(ctx, ` +insert into sync_attempt_failures(repo_id, thread_id, number, operation, error_class, error_message, first_seen_at, last_seen_at, retry_count, resolved_at) +values(?, ?, ?, ?, ?, ?, ?, ?, 0, null) +on conflict(repo_id, number, operation, error_class) do update set + thread_id=coalesce(excluded.thread_id, sync_attempt_failures.thread_id), + error_message=excluded.error_message, + last_seen_at=excluded.last_seen_at, + retry_count=sync_attempt_failures.retry_count + 1, + resolved_at=null +`, failure.RepoID, threadID, failure.Number, operation, errorClass, message, firstSeen, lastSeen); err != nil { + return 0, fmt.Errorf("record sync attempt failure: %w", err) + } + var id int64 + if err := s.q().QueryRowContext(ctx, ` +select id +from sync_attempt_failures +where repo_id = ? and number = ? and operation = ? and error_class = ? +`, failure.RepoID, failure.Number, operation, errorClass).Scan(&id); err != nil { + return 0, fmt.Errorf("read sync attempt failure id: %w", err) + } + return id, nil +} + +func (s *Store) ResolveSyncAttemptFailures(ctx context.Context, repoID int64, number int, resolvedAt string) (int, error) { + if repoID == 0 || number <= 0 { + return 0, nil + } + result, err := s.q().ExecContext(ctx, ` +update sync_attempt_failures +set resolved_at = ?, last_seen_at = ? +where repo_id = ? + and number = ? + and resolved_at is null +`, resolvedAt, resolvedAt, repoID, number) + if err != nil { + return 0, fmt.Errorf("resolve sync attempt failures: %w", err) + } + affected, err := result.RowsAffected() + if err != nil { + return 0, fmt.Errorf("count resolved sync attempt failures: %w", err) + } + return int(affected), nil +} + +func (s *Store) ListSyncAttemptFailures(ctx context.Context, options SyncAttemptFailureListOptions) ([]SyncAttemptFailure, error) { + if options.RepoID == 0 { + return nil, fmt.Errorf("list sync attempt failures: missing repo id") + } + limit := options.Limit + if limit <= 0 { + limit = 50 + } + whereResolved := "and resolved_at is null" + if options.IncludeResolved { + whereResolved = "" + } + rows, err := s.q().QueryContext(ctx, ` +select id, repo_id, thread_id, number, operation, error_class, error_message, first_seen_at, last_seen_at, retry_count, resolved_at +from sync_attempt_failures +where repo_id = ? +`+whereResolved+` +order by resolved_at is not null, last_seen_at desc, id desc +limit ?`, options.RepoID, limit) + if err != nil { + return nil, fmt.Errorf("list sync attempt failures: %w", err) + } + defer rows.Close() + var out []SyncAttemptFailure + for rows.Next() { + var failure SyncAttemptFailure + var threadID sql.NullInt64 + var resolvedAt sql.NullString + if err := rows.Scan( + &failure.ID, + &failure.RepoID, + &threadID, + &failure.Number, + &failure.Operation, + &failure.ErrorClass, + &failure.ErrorMessage, + &failure.FirstSeenAt, + &failure.LastSeenAt, + &failure.RetryCount, + &resolvedAt, + ); err != nil { + return nil, fmt.Errorf("scan sync attempt failure: %w", err) + } + if threadID.Valid { + failure.ThreadID = threadID.Int64 + } + if resolvedAt.Valid { + failure.ResolvedAt = resolvedAt.String + } + out = append(out, failure) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("scan sync attempt failures: %w", err) + } + return out, nil +} diff --git a/internal/store/sync_failures_test.go b/internal/store/sync_failures_test.go new file mode 100644 index 0000000..45f56e4 --- /dev/null +++ b/internal/store/sync_failures_test.go @@ -0,0 +1,90 @@ +package store + +import ( + "context" + "path/filepath" + "testing" +) + +func TestSyncAttemptFailureRetryAndResolve(t *testing.T) { + ctx := context.Background() + st, err := Open(ctx, filepath.Join(t.TempDir(), "gitcrawl.db")) + if err != nil { + t.Fatalf("open store: %v", err) + } + defer st.Close() + + repoID, err := st.UpsertRepository(ctx, Repository{Owner: "openclaw", Name: "gitcrawl", FullName: "openclaw/gitcrawl", RawJSON: "{}", UpdatedAt: "2026-06-06T00:00:00Z"}) + if err != nil { + t.Fatalf("repo: %v", err) + } + threadID, err := st.UpsertThread(ctx, Thread{ + RepoID: repoID, GitHubID: "84", Number: 84, Kind: "pull_request", State: "open", + Title: "Track failed sync attempts", HTMLURL: "https://github.com/openclaw/gitcrawl/pull/84", + LabelsJSON: "[]", AssigneesJSON: "[]", RawJSON: "{}", ContentHash: "h84", UpdatedAt: "2026-06-06T00:00:00Z", + }) + if err != nil { + t.Fatalf("thread: %v", err) + } + first := SyncAttemptFailure{ + RepoID: repoID, ThreadID: threadID, Number: 84, Operation: "pull_request_details", ErrorClass: "rate_limit", + ErrorMessage: "secondary rate limit", FirstSeenAt: "2026-06-06T00:00:00Z", LastSeenAt: "2026-06-06T00:00:00Z", + } + id, err := st.RecordSyncAttemptFailure(ctx, first) + if err != nil { + t.Fatalf("record first failure: %v", err) + } + if _, err := st.RecordSyncAttemptFailure(ctx, SyncAttemptFailure{ + RepoID: repoID, ThreadID: threadID, Number: 84, Operation: "pull_request_details", ErrorClass: "rate_limit", + ErrorMessage: "secondary rate limit again", LastSeenAt: "2026-06-06T00:05:00Z", + }); err != nil { + t.Fatalf("record retry failure: %v", err) + } + + failures, err := st.ListSyncAttemptFailures(ctx, SyncAttemptFailureListOptions{RepoID: repoID}) + if err != nil { + t.Fatalf("list failures: %v", err) + } + if len(failures) != 1 { + t.Fatalf("failure count = %d, want 1", len(failures)) + } + if failures[0].ID != id || failures[0].RetryCount != 1 || failures[0].ResolvedAt != "" || failures[0].ErrorMessage != "secondary rate limit again" { + t.Fatalf("failure = %+v", failures[0]) + } + + resolved, err := st.ResolveSyncAttemptFailures(ctx, repoID, 84, "2026-06-06T00:10:00Z") + if err != nil { + t.Fatalf("resolve failures: %v", err) + } + if resolved != 1 { + t.Fatalf("resolved = %d, want 1", resolved) + } + unresolved, err := st.ListSyncAttemptFailures(ctx, SyncAttemptFailureListOptions{RepoID: repoID}) + if err != nil { + t.Fatalf("list unresolved failures: %v", err) + } + if len(unresolved) != 0 { + t.Fatalf("unresolved failures = %+v", unresolved) + } + history, err := st.ListSyncAttemptFailures(ctx, SyncAttemptFailureListOptions{RepoID: repoID, IncludeResolved: true}) + if err != nil { + t.Fatalf("list history failures: %v", err) + } + if len(history) != 1 || history[0].ResolvedAt != "2026-06-06T00:10:00Z" { + t.Fatalf("history = %+v", history) + } + + if _, err := st.RecordSyncAttemptFailure(ctx, SyncAttemptFailure{ + RepoID: repoID, ThreadID: threadID, Number: 84, Operation: "pull_request_details", ErrorClass: "rate_limit", + ErrorMessage: "secondary rate limit after resolve", LastSeenAt: "2026-06-06T00:15:00Z", + }); err != nil { + t.Fatalf("record unresolved retry: %v", err) + } + failures, err = st.ListSyncAttemptFailures(ctx, SyncAttemptFailureListOptions{RepoID: repoID}) + if err != nil { + t.Fatalf("list reopened failures: %v", err) + } + if len(failures) != 1 || failures[0].RetryCount != 2 || failures[0].ResolvedAt != "" { + t.Fatalf("reopened failure = %+v", failures) + } +} diff --git a/internal/syncer/syncer.go b/internal/syncer/syncer.go index fcd8bfb..37dca7c 100644 --- a/internal/syncer/syncer.go +++ b/internal/syncer/syncer.go @@ -5,6 +5,7 @@ import ( "crypto/sha256" "encoding/hex" "encoding/json" + "errors" "fmt" "log/slog" "strconv" @@ -156,12 +157,18 @@ 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) + } return Stats{}, err } 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) + } return Stats{}, err } payload.pullDetails = pullDetails @@ -242,6 +249,9 @@ func (s *Syncer) Sync(ctx context.Context, options Options) (Stats, error) { if err != nil { return err } + if _, err := st.ResolveSyncAttemptFailures(ctx, repoID, thread.Number, s.now().Format(time.RFC3339Nano)); err != nil { + return err + } attempt.PRDetailsSynced++ attempt.PRFilesSynced += detailStats.files attempt.PRCommitsSynced += detailStats.commits @@ -309,6 +319,53 @@ func (s *Syncer) Sync(ctx context.Context, options Options) (Stats, error) { return stats, nil } +func (s *Syncer) recordPullRequestSyncFailure(ctx context.Context, options Options, repoRaw, row map[string]any, operation string, syncErr error) error { + if syncErr == nil { + return nil + } + return s.store.WithTx(ctx, func(st *store.Store) error { + now := s.now().Format(time.RFC3339Nano) + 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: now, + }) + if err != nil { + return err + } + thread := mapIssueToThread(repoID, row, now) + threadID, err := st.UpsertThread(ctx, thread) + if err != nil { + return err + } + _, err = st.RecordSyncAttemptFailure(ctx, store.SyncAttemptFailure{ + RepoID: repoID, + ThreadID: threadID, + Number: thread.Number, + Operation: operation, + ErrorClass: syncAttemptErrorClass(syncErr), + ErrorMessage: syncErr.Error(), + FirstSeenAt: now, + LastSeenAt: now, + }) + return err + }) +} + +func syncAttemptErrorClass(err error) string { + switch { + case errors.Is(err, context.Canceled): + return "context_canceled" + case errors.Is(err, context.DeadlineExceeded): + return "deadline_exceeded" + default: + return "error" + } +} + func applySyncPersistStats(stats *Stats, persisted syncPersistStats) { stats.ThreadsSynced = persisted.ThreadsSynced stats.IssuesSynced = persisted.IssuesSynced diff --git a/internal/syncer/syncer_test.go b/internal/syncer/syncer_test.go index 69c92a7..df8ed8e 100644 --- a/internal/syncer/syncer_test.go +++ b/internal/syncer/syncer_test.go @@ -680,13 +680,42 @@ func TestSyncPullRequestDetailsFailsOnReviewThreadFetchError(t *testing.T) { if err == nil || !strings.Contains(err.Error(), "list pull request review threads for #8") { t.Fatalf("sync error = %v", err) } - if _, err := st.RepositoryByFullName(ctx, "openclaw/gitcrawl"); err == nil { - t.Fatal("repository persisted after failed PR detail hydration") + repo, err := st.RepositoryByFullName(ctx, "openclaw/gitcrawl") + if err != nil { + t.Fatalf("repository not persisted for failed PR detail hydration: %v", err) + } + threads, err := st.ListThreads(ctx, repo.ID, true) + if err != nil { + t.Fatalf("threads: %v", err) + } + if len(threads) != 1 || threads[0].Number != 8 || threads[0].Kind != "pull_request" { + t.Fatalf("threads = %+v", threads) + } + failures, err := st.ListSyncAttemptFailures(ctx, store.SyncAttemptFailureListOptions{RepoID: repo.ID}) + if err != nil { + t.Fatalf("sync attempt failures: %v", err) + } + if len(failures) != 1 { + t.Fatalf("failure count = %d, want 1", len(failures)) + } + if failures[0].Number != 8 || failures[0].Operation != "pull_review_threads" || !strings.Contains(failures[0].ErrorMessage, "graphql unavailable") || failures[0].ResolvedAt != "" { + t.Fatalf("failure = %+v", failures[0]) } - assertTableRowCount(t, st, "repositories", 0) - assertTableRowCount(t, st, "threads", 0) assertTableRowCount(t, st, "pull_request_review_thread_syncs", 0) assertTableRowCount(t, st, "sync_runs", 0) + + s = New(pullDetailsGitHub{}, st) + s.now = func() time.Time { return time.Date(2026, 4, 26, 0, 1, 0, 0, time.UTC) } + if _, err := s.Sync(ctx, Options{Owner: "openclaw", Repo: "gitcrawl", Numbers: []int{8}, IncludePRDetails: true}); err != nil { + t.Fatalf("sync retry: %v", err) + } + history, err := st.ListSyncAttemptFailures(ctx, store.SyncAttemptFailureListOptions{RepoID: repo.ID, IncludeResolved: true}) + if err != nil { + t.Fatalf("sync attempt failure history: %v", err) + } + if len(history) != 1 || history[0].ResolvedAt == "" { + t.Fatalf("history = %+v", history) + } } func TestSyncPullRequestDetailsSkipsCheckAndWorkflowFetchWithoutHeadSHA(t *testing.T) { From 62069ed367827644fe1e6aebeb3e2fa2d9c948b0 Mon Sep 17 00:00:00 2001 From: Andy Ye <35905412+TurboTheTurtle@users.noreply.github.com> Date: Wed, 1 Jul 2026 12:42:02 -0700 Subject: [PATCH 2/3] Handle pre-ledger sync failure databases --- internal/store/sync_failures.go | 12 ++++++++ internal/store/sync_failures_test.go | 44 ++++++++++++++++++++++++++++ 2 files changed, 56 insertions(+) diff --git a/internal/store/sync_failures.go b/internal/store/sync_failures.go index 8677018..d832367 100644 --- a/internal/store/sync_failures.go +++ b/internal/store/sync_failures.go @@ -7,6 +7,8 @@ import ( "strings" ) +const syncAttemptFailuresSchemaVersion = 5 + type SyncAttemptFailure struct { ID int64 `json:"id"` RepoID int64 `json:"repo_id"` @@ -106,6 +108,16 @@ func (s *Store) ListSyncAttemptFailures(ctx context.Context, options SyncAttempt if options.RepoID == 0 { return nil, fmt.Errorf("list sync attempt failures: missing repo id") } + if !s.hasTable(ctx, "sync_attempt_failures") { + current, err := s.schemaVersion(ctx) + if err != nil { + return nil, fmt.Errorf("list sync attempt failures: %w", err) + } + if current < syncAttemptFailuresSchemaVersion { + return []SyncAttemptFailure{}, nil + } + return nil, fmt.Errorf("list sync attempt failures: missing sync_attempt_failures table") + } limit := options.Limit if limit <= 0 { limit = 50 diff --git a/internal/store/sync_failures_test.go b/internal/store/sync_failures_test.go index 45f56e4..183f079 100644 --- a/internal/store/sync_failures_test.go +++ b/internal/store/sync_failures_test.go @@ -2,6 +2,7 @@ package store import ( "context" + "database/sql" "path/filepath" "testing" ) @@ -88,3 +89,46 @@ func TestSyncAttemptFailureRetryAndResolve(t *testing.T) { t.Fatalf("reopened failure = %+v", failures) } } + +func TestListSyncAttemptFailuresTreatsPreLedgerReadOnlyStoreAsEmpty(t *testing.T) { + ctx := context.Background() + dbPath := filepath.Join(t.TempDir(), "gitcrawl.db") + st, err := Open(ctx, dbPath) + if err != nil { + t.Fatalf("open store: %v", err) + } + repoID, err := st.UpsertRepository(ctx, Repository{Owner: "openclaw", Name: "gitcrawl", FullName: "openclaw/gitcrawl", RawJSON: "{}", UpdatedAt: "2026-06-06T00:00:00Z"}) + if err != nil { + t.Fatalf("repo: %v", err) + } + if err := st.Close(); err != nil { + t.Fatalf("close store: %v", err) + } + + db, err := sql.Open("sqlite", dbPath) + if err != nil { + t.Fatalf("open raw db: %v", err) + } + if _, err := db.ExecContext(ctx, ` +drop table sync_attempt_failures; +pragma user_version = 4; +`); err != nil { + t.Fatalf("downgrade ledger schema: %v", err) + } + if err := db.Close(); err != nil { + t.Fatalf("close raw db: %v", err) + } + + readOnly, err := OpenReadOnly(ctx, dbPath) + if err != nil { + t.Fatalf("open read-only store: %v", err) + } + defer readOnly.Close() + failures, err := readOnly.ListSyncAttemptFailures(ctx, SyncAttemptFailureListOptions{RepoID: repoID}) + if err != nil { + t.Fatalf("list failures: %v", err) + } + if len(failures) != 0 { + t.Fatalf("failures = %+v, want empty", failures) + } +} From 055cbf7ef546e7530d41857d513ce1445f715ec3 Mon Sep 17 00:00:00 2001 From: Andy Ye <35905412+TurboTheTurtle@users.noreply.github.com> Date: Tue, 7 Jul 2026 11:10:21 -0700 Subject: [PATCH 3/3] test: cover sync failure helpers --- internal/cli/gh_compat_helpers_test.go | 66 ++++++++++++++++++++++++++ internal/syncer/syncer_test.go | 32 +++++++++++++ 2 files changed, 98 insertions(+) diff --git a/internal/cli/gh_compat_helpers_test.go b/internal/cli/gh_compat_helpers_test.go index 081219c..b2ca35b 100644 --- a/internal/cli/gh_compat_helpers_test.go +++ b/internal/cli/gh_compat_helpers_test.go @@ -2,10 +2,14 @@ package cli import ( "bytes" + "errors" + "fmt" "os" "path/filepath" "strings" "testing" + + "github.com/openclaw/gitcrawl/internal/store" ) func TestGHAPIRouteNormalization(t *testing.T) { @@ -76,3 +80,65 @@ func TestWriteJSONValueWithoutJQ(t *testing.T) { t.Fatalf("json output = %s", out.String()) } } + +func TestWriteJSONValueErrors(t *testing.T) { + app := New() + if err := app.writeJSONValue(map[string]any{"bad": func() {}}, ""); err == nil { + t.Fatalf("expected marshal error") + } + t.Setenv("PATH", "") + if err := app.writeJSONValue(map[string]any{"ok": true}, ".ok"); !errors.Is(err, errLocalGHUnsupported) { + t.Fatalf("jq error = %v, want local gh unsupported", err) + } +} + +func TestLocalGHUnsupportedAndShim(t *testing.T) { + if err := localGHUnsupported(nil); !errors.Is(err, errLocalGHUnsupported) { + t.Fatalf("nil wrapped error = %v", err) + } + if err := localGHUnsupported(errors.New("missing")); !errors.Is(err, errLocalGHUnsupported) || !strings.Contains(err.Error(), "missing") { + t.Fatalf("wrapped error = %v", err) + } + + app := New() + var stderr bytes.Buffer + app.Stderr = &stderr + err := app.runGHShim(nil, nil) + if err == nil || !strings.Contains(err.Error(), "gitcrawl gh moved to octopool") { + t.Fatalf("runGHShim error = %v", err) + } + if !strings.Contains(stderr.String(), "octopool login") { + t.Fatalf("stderr = %s", stderr.String()) + } +} + +func TestClusterAndRuntimeHelperBranches(t *testing.T) { + clusterCases := []struct { + source string + want string + ok bool + }{ + {source: "", want: "", ok: true}, + {source: " AUTO ", want: "", ok: true}, + {source: "raw", want: store.ClusterSourceRun, ok: true}, + {source: "durable", want: store.ClusterSourceDurable, ok: true}, + {source: "unknown", ok: false}, + } + for _, tc := range clusterCases { + got, err := parseClusterDetailSource(tc.source) + if tc.ok && (err != nil || got != tc.want) { + t.Fatalf("parseClusterDetailSource(%q) = %q, %v; want %q", tc.source, got, err, tc.want) + } + if !tc.ok && err == nil { + t.Fatalf("parseClusterDetailSource(%q) unexpectedly succeeded", tc.source) + } + } + + err := neighborEmbeddingRecoveryError("openclaw", "gitcrawl", 42, true, fmt.Errorf("missing vector")) + if !strings.Contains(err.Error(), "gitcrawl embed openclaw/gitcrawl --number 42 --limit 1 --include-closed") { + t.Fatalf("neighbor recovery error = %v", err) + } + if isGitIndexLockError(nil) || isGitIndexLockError(errors.New("index.lock")) || !isGitIndexLockError(errors.New("fatal: Unable to create '.git/index.lock': File exists.")) { + t.Fatalf("index lock classifier failed") + } +} diff --git a/internal/syncer/syncer_test.go b/internal/syncer/syncer_test.go index df8ed8e..95ad88a 100644 --- a/internal/syncer/syncer_test.go +++ b/internal/syncer/syncer_test.go @@ -718,6 +718,38 @@ func TestSyncPullRequestDetailsFailsOnReviewThreadFetchError(t *testing.T) { } } +func TestSyncAttemptErrorClass(t *testing.T) { + tests := []struct { + name string + err error + want string + }{ + { + name: "context canceled", + err: context.Canceled, + want: "context_canceled", + }, + { + name: "deadline exceeded", + err: context.DeadlineExceeded, + want: "deadline_exceeded", + }, + { + name: "generic error", + err: errors.New("graphql unavailable"), + want: "error", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := syncAttemptErrorClass(tt.err); got != tt.want { + t.Fatalf("syncAttemptErrorClass() = %q, want %q", got, tt.want) + } + }) + } +} + func TestSyncPullRequestDetailsSkipsCheckAndWorkflowFetchWithoutHeadSHA(t *testing.T) { ctx := context.Background() st, err := store.Open(ctx, filepath.Join(t.TempDir(), "gitcrawl.db"))