From 258773618f9bccc2cb3aadcdf79093bf20f3a163 Mon Sep 17 00:00:00 2001 From: espadonne Date: Thu, 28 May 2026 02:10:11 -0400 Subject: [PATCH 1/2] Bound lifecycle sweep hard deletes --- docs/internal/repo-lifecycle.md | 10 ++- internal/repos/lifecycle/hard_delete.go | 5 +- internal/worker/jobs/lifecycle_sweep.go | 87 +++++++++++++++----- internal/worker/jobs/lifecycle_sweep_test.go | 56 +++++++++++++ 4 files changed, 135 insertions(+), 23 deletions(-) create mode 100644 internal/worker/jobs/lifecycle_sweep_test.go diff --git a/docs/internal/repo-lifecycle.md b/docs/internal/repo-lifecycle.md index d57de226..a273ee9d 100644 --- a/docs/internal/repo-lifecycle.md +++ b/docs/internal/repo-lifecycle.md @@ -101,7 +101,10 @@ data layer; visibility flips don't cascade. * **Hard delete**: `lifecycle:sweep` worker job (registered in `cmd/shithubd/worker.go`) runs periodically. The handler: 1. `ListRepoIDsPastSoftDeleteGrace` — finds rows past 7 days. - 2. For each, `lifecycle.HardDelete` runs: + 2. For each, `lifecycle.HardDelete` runs under a 45-second child + timeout. A timeout or other per-repo failure is logged and the + sweep continues with later tombstones so one stuck row cannot + consume the whole worker job timeout. * `OrphanForksOf(repoID)` — children's `fork_of_repo_id` set NULL. * `DELETE FROM repos WHERE id=$1`. FK ON DELETE CASCADE handles `push_events`, `repo_collaborators`, `repo_redirects` (rows @@ -142,6 +145,11 @@ from "redirected" to "never existed." rather than the active canonical path. If disk pressure becomes an issue, configure the grace window down (it's a constant in `lifecycle.go::softDeleteGrace`; promote to config in S37 if needed). +* Repeated `lifecycle:sweep` failures for the same `repo_id` usually + indicate a stuck hard-delete row, a slow filesystem tombstone delete, + or a database wait while taking the owner/name advisory lock. Later + rows should still progress because each hard-delete gets its own + timeout budget. * Renaming a repo doesn't require restarting any worker or hook. The atomic FS move + DB update means the next hook invocation will resolve the new path. diff --git a/internal/repos/lifecycle/hard_delete.go b/internal/repos/lifecycle/hard_delete.go index b5953334..6b000586 100644 --- a/internal/repos/lifecycle/hard_delete.go +++ b/internal/repos/lifecycle/hard_delete.go @@ -5,6 +5,7 @@ package lifecycle import ( "context" "fmt" + "time" "github.com/jackc/pgx/v5/pgtype" @@ -44,7 +45,9 @@ func HardDelete(ctx context.Context, deps Deps, actorUserID, repoID int64) error committed := false defer func() { if !committed { - _ = tx.Rollback(ctx) + rollbackCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = tx.Rollback(rollbackCtx) } }() diff --git a/internal/worker/jobs/lifecycle_sweep.go b/internal/worker/jobs/lifecycle_sweep.go index 17fe6450..b2bdfbb1 100644 --- a/internal/worker/jobs/lifecycle_sweep.go +++ b/internal/worker/jobs/lifecycle_sweep.go @@ -6,6 +6,7 @@ import ( "context" "encoding/json" "log/slog" + "time" "github.com/jackc/pgx/v5/pgxpool" @@ -17,6 +18,8 @@ import ( "github.com/tenseleyFlow/shithub/internal/worker" ) +const lifecycleSweepHardDeleteTimeout = 45 * time.Second + // LifecycleSweepDeps wires the periodic sweep handler. type LifecycleSweepDeps struct { Pool *pgxpool.Pool @@ -29,9 +32,9 @@ type LifecycleSweepDeps struct { // // 1. Hard-delete every repo whose deleted_at is past the grace window. // Each repo gets the full lifecycle.HardDelete cascade — FS + DB -// + audit. We process inline rather than fanning out to one job -// per repo because hard-deletes are rare and the per-row cost is -// small. +// + audit. Each row gets a bounded child context so one stuck +// tombstone cannot consume the entire worker job timeout and starve +// later rows. // 2. Flip pending transfer requests past expires_at to "expired". // // Enqueue this kind from a cron timer (S37 ships the systemd cron @@ -49,15 +52,10 @@ func LifecycleSweep(deps LifecycleSweepDeps) worker.Handler { Pool: deps.Pool, RepoFS: deps.RepoFS, Audit: deps.Audit, Logger: deps.Logger, } - for _, id := range ids { - if err := ctx.Err(); err != nil { - return err - } - if err := lifecycle.HardDelete(ctx, ldeps, 0, id); err != nil { - deps.Logger.WarnContext(ctx, "lifecycle:sweep: hard delete failed", - "repo_id", id, "error", err) - // Keep going — one bad row shouldn't poison the sweep. - } + if err := runLifecycleSweepHardDeletes(ctx, ids, lifecycleSweepHardDeleteTimeout, deps.Logger, func(deleteCtx context.Context, id int64) error { + return lifecycle.HardDelete(deleteCtx, ldeps, 0, id) + }); err != nil { + return err } // 2. Expire pending transfers past their TTL. @@ -77,17 +75,64 @@ func LifecycleSweep(deps LifecycleSweepDeps) worker.Handler { ohd := orgs.HardDeleteDeps{Deps: odeps, RepoFS: deps.RepoFS, Audit: deps.Audit} orgIDs, err := orgs.ListPastGraceOrgIDs(ctx, odeps) if err != nil { - deps.Logger.WarnContext(ctx, "lifecycle:sweep: list past-grace orgs", "error", err) + warnLifecycleSweep(ctx, deps.Logger, "lifecycle:sweep: list past-grace orgs", "error", err) } - for _, oid := range orgIDs { - if err := ctx.Err(); err != nil { - return err - } - if err := orgs.HardDelete(ctx, ohd, oid); err != nil { - deps.Logger.WarnContext(ctx, "lifecycle:sweep: org hard delete failed", - "org_id", oid, "error", err) - } + if err := runLifecycleSweepOrgHardDeletes(ctx, orgIDs, lifecycleSweepHardDeleteTimeout, deps.Logger, func(deleteCtx context.Context, id int64) error { + return orgs.HardDelete(deleteCtx, ohd, id) + }); err != nil { + return err } return nil } } + +func runLifecycleSweepHardDeletes( + ctx context.Context, + ids []int64, + timeout time.Duration, + logger *slog.Logger, + hardDelete func(context.Context, int64) error, +) error { + for _, id := range ids { + if err := ctx.Err(); err != nil { + return err + } + deleteCtx, cancel := context.WithTimeout(ctx, timeout) + err := hardDelete(deleteCtx, id) + cancel() + if err != nil { + warnLifecycleSweep(ctx, logger, "lifecycle:sweep: hard delete failed", + "repo_id", id, "error", err) + // Keep going — one bad row shouldn't poison the sweep. + } + } + return nil +} + +func runLifecycleSweepOrgHardDeletes( + ctx context.Context, + ids []int64, + timeout time.Duration, + logger *slog.Logger, + hardDelete func(context.Context, int64) error, +) error { + for _, id := range ids { + if err := ctx.Err(); err != nil { + return err + } + deleteCtx, cancel := context.WithTimeout(ctx, timeout) + err := hardDelete(deleteCtx, id) + cancel() + if err != nil { + warnLifecycleSweep(ctx, logger, "lifecycle:sweep: org hard delete failed", + "org_id", id, "error", err) + } + } + return nil +} + +func warnLifecycleSweep(ctx context.Context, logger *slog.Logger, msg string, args ...any) { + if logger != nil { + logger.WarnContext(ctx, msg, args...) + } +} diff --git a/internal/worker/jobs/lifecycle_sweep_test.go b/internal/worker/jobs/lifecycle_sweep_test.go new file mode 100644 index 00000000..11fd166e --- /dev/null +++ b/internal/worker/jobs/lifecycle_sweep_test.go @@ -0,0 +1,56 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package jobs + +import ( + "context" + "errors" + "io" + "log/slog" + "testing" + "time" +) + +func TestRunLifecycleSweepHardDeletesBoundsStuckRows(t *testing.T) { + t.Parallel() + + var seen []int64 + err := runLifecycleSweepHardDeletes( + context.Background(), + []int64{10, 20}, + 5*time.Millisecond, + slog.New(slog.NewTextHandler(io.Discard, nil)), + func(ctx context.Context, id int64) error { + seen = append(seen, id) + if id == 10 { + <-ctx.Done() + return ctx.Err() + } + return nil + }, + ) + if err != nil { + t.Fatalf("runLifecycleSweepHardDeletes: %v", err) + } + if len(seen) != 2 || seen[0] != 10 || seen[1] != 20 { + t.Fatalf("seen ids = %v, want [10 20]", seen) + } +} + +func TestRunLifecycleSweepHardDeletesStopsWhenParentCanceled(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + called := false + err := runLifecycleSweepHardDeletes(ctx, []int64{10}, time.Second, nil, func(context.Context, int64) error { + called = true + return nil + }) + if !errors.Is(err, context.Canceled) { + t.Fatalf("err = %v, want context.Canceled", err) + } + if called { + t.Fatal("hard delete called after parent context was canceled") + } +} From f35ddf77b00f575582f9a1fc8bdc87cbb8935142 Mon Sep 17 00:00:00 2001 From: espadonne Date: Thu, 28 May 2026 02:11:11 -0400 Subject: [PATCH 2/2] Prefilter code index blobs by size --- docs/internal/search.md | 4 + internal/repos/git/treeops.go | 48 ++++++++ internal/repos/git/treeops_test.go | 37 +++++++ internal/worker/jobs/repo_index_code.go | 23 ++-- internal/worker/jobs/repo_index_code_test.go | 111 +++++++++++++++++++ 5 files changed, 214 insertions(+), 9 deletions(-) create mode 100644 internal/worker/jobs/repo_index_code_test.go diff --git a/docs/internal/search.md b/docs/internal/search.md index 293652b7..ec56ed32 100644 --- a/docs/internal/search.md +++ b/docs/internal/search.md @@ -193,6 +193,10 @@ invalid `state:all` query text. a push advances the repo's default branch. The job is idempotent + atomic-swap, so concurrent pushes that land while the previous index is running re-trigger on the next push tick. +* **Tree walk**: the worker uses one `git ls-tree -r --long -z` pass + to collect paths and blob sizes. Paths for oversize files are kept, + but the worker does not spawn `git cat-file` for blobs that already + exceed the content-index size gate. * **Atomic swap**: the worker runs `DELETE … + INSERT …` for the repo in one tx. Readers never see a partial index. * **Size + textness gates**: diff --git a/internal/repos/git/treeops.go b/internal/repos/git/treeops.go index 4b91e292..ed1d61cb 100644 --- a/internal/repos/git/treeops.go +++ b/internal/repos/git/treeops.go @@ -129,6 +129,14 @@ type TreeBlob struct { Size int64 } +// TreeFile is one regular file-like entry returned by a recursive tree +// walk. It includes regular blobs and symlinks, but excludes trees and +// submodule commit entries. +type TreeFile struct { + Path string + Size int64 +} + // LsTree lists entries at :. Empty path lists the repo root. // Returns an empty slice when the path doesn't exist or is itself a // blob — callers should fall back to BlobInfo. @@ -227,6 +235,46 @@ func ListBlobs(ctx context.Context, gitDir, ref string) ([]TreeBlob, error) { return blobs, nil } +// ListFiles returns every regular file-like path under ref with its +// git-reported byte size. Unlike ListBlobs, it keeps symlinks because +// code search should still index the symlink path. +func ListFiles(ctx context.Context, gitDir, ref string) ([]TreeFile, error) { + cmd := exec.CommandContext(ctx, "git", "-C", gitDir, + "ls-tree", "-r", "--long", "--full-tree", "-z", ref) + out, err := cmd.Output() + if err != nil { + return nil, wrapExecErr(err) + } + + records := bytes.Split(bytes.TrimRight(out, "\x00"), []byte{0}) + files := make([]TreeFile, 0, len(records)) + for _, rec := range records { + if len(rec) == 0 { + continue + } + tabIdx := bytes.IndexByte(rec, '\t') + if tabIdx < 0 { + continue + } + left, name := string(rec[:tabIdx]), string(rec[tabIdx+1:]) + fields := strings.Fields(left) + if len(fields) != 4 { + continue + } + kind := classifyEntry(fields[0], fields[1]) + if kind != EntryBlob && kind != EntrySymlink { + continue + } + size, err := strconv.ParseInt(fields[3], 10, 64) + if err != nil { + continue + } + files = append(files, TreeFile{Path: name, Size: size}) + } + sort.Slice(files, func(i, j int) bool { return files[i].Path < files[j].Path }) + return files, nil +} + // classifyEntry maps git's mode+type fields to our four kinds. // Symlinks come in with mode 120000 type=blob; we surface them as // symlink so the UI can avoid Reading them. diff --git a/internal/repos/git/treeops_test.go b/internal/repos/git/treeops_test.go index 08020413..e40dab73 100644 --- a/internal/repos/git/treeops_test.go +++ b/internal/repos/git/treeops_test.go @@ -105,3 +105,40 @@ func TestListBlobs_RecursiveSizes(t *testing.T) { t.Errorf("readme blob size = %d", got["README.md"]) } } + +func TestListFiles_RecursiveSizes(t *testing.T) { + t.Parallel() + dir := t.TempDir() + if out, err := exec.Command("git", "init", "--bare", "--initial-branch=trunk", dir).CombinedOutput(); err != nil { + t.Fatalf("git init: %v: %s", err, out) + } + build := gitops.InitialCommit{ + GitDir: dir, + AuthorName: "Test Author", + AuthorEmail: "test@example.com", + Branch: "trunk", + When: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC), + Files: []gitops.FileEntry{ + {Path: "README.md", Body: []byte("# demo\n")}, + {Path: "big.txt", Body: []byte("0123456789")}, + }, + } + if _, err := build.Build(context.Background()); err != nil { + t.Fatalf("InitialCommit.Build: %v", err) + } + + files, err := gitops.ListFiles(context.Background(), dir, "trunk") + if err != nil { + t.Fatalf("ListFiles: %v", err) + } + got := map[string]int64{} + for _, f := range files { + got[f.Path] = f.Size + } + if got["README.md"] != int64(len("# demo\n")) { + t.Errorf("readme size = %d", got["README.md"]) + } + if got["big.txt"] != int64(len("0123456789")) { + t.Errorf("big size = %d", got["big.txt"]) + } +} diff --git a/internal/worker/jobs/repo_index_code.go b/internal/worker/jobs/repo_index_code.go index ad9d204b..9f39232b 100644 --- a/internal/worker/jobs/repo_index_code.go +++ b/internal/worker/jobs/repo_index_code.go @@ -102,8 +102,10 @@ func RepoIndexCode(deps IndexCodeDeps) worker.Handler { return fmt.Errorf("resolve %s: %w", ref, err) } - // Walk the tree. - paths, err := repogit.ListAllPaths(ctx, gitDir, ref) + // Walk the tree once with git-reported sizes. Oversize blobs + // still get path rows, but we do not spawn cat-file just to + // rediscover that their content is too large to index. + files, err := repogit.ListFiles(ctx, gitDir, ref) if err != nil { return fmt.Errorf("ls-tree: %w", err) } @@ -114,18 +116,21 @@ func RepoIndexCode(deps IndexCodeDeps) worker.Handler { path string content []byte // empty if skipped from content index } - entries := make([]indexed, 0, len(paths)) - for _, path := range paths { + entries := make([]indexed, 0, len(files)) + for _, file := range files { + path := file.Path if shouldSkipPath(path) { continue } ent := indexed{path: path} - blob, err := repogit.ReadBlobBytes(ctx, gitDir, ref, path, maxFileBytes+1) - if err == nil && len(blob) <= maxFileBytes && isText(blob) { - if len(blob) > maxIndexBytes { - blob = blob[:maxIndexBytes] + if file.Size <= maxFileBytes { + blob, err := repogit.ReadBlobBytes(ctx, gitDir, ref, path, maxFileBytes) + if err == nil && isText(blob) { + if len(blob) > maxIndexBytes { + blob = blob[:maxIndexBytes] + } + ent.content = blob } - ent.content = blob } entries = append(entries, ent) } diff --git a/internal/worker/jobs/repo_index_code_test.go b/internal/worker/jobs/repo_index_code_test.go new file mode 100644 index 00000000..24efc1af --- /dev/null +++ b/internal/worker/jobs/repo_index_code_test.go @@ -0,0 +1,111 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package jobs_test + +import ( + "context" + "encoding/json" + "io" + "log/slog" + "strings" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgtype" + + "github.com/tenseleyFlow/shithub/internal/infra/storage" + repogit "github.com/tenseleyFlow/shithub/internal/repos/git" + reposdb "github.com/tenseleyFlow/shithub/internal/repos/sqlc" + "github.com/tenseleyFlow/shithub/internal/testing/dbtest" + usersdb "github.com/tenseleyFlow/shithub/internal/users/sqlc" + "github.com/tenseleyFlow/shithub/internal/worker/jobs" +) + +func TestRepoIndexCodeSkipsOversizeBlobContent(t *testing.T) { + t.Parallel() + ctx := context.Background() + pool := dbtest.NewTestDB(t) + rfs, err := storage.NewRepoFS(t.TempDir()) + if err != nil { + t.Fatalf("NewRepoFS: %v", err) + } + + user, err := usersdb.New().CreateUser(ctx, pool, usersdb.CreateUserParams{ + Username: "indexer", DisplayName: "Indexer", PasswordHash: fixtureHash, + }) + if err != nil { + t.Fatalf("CreateUser: %v", err) + } + rq := reposdb.New() + repo, err := rq.CreateRepo(ctx, pool, reposdb.CreateRepoParams{ + OwnerUserID: pgtype.Int8{Int64: user.ID, Valid: true}, + Name: "indexed", + DefaultBranch: "trunk", + Visibility: reposdb.RepoVisibilityPublic, + }) + if err != nil { + t.Fatalf("CreateRepo: %v", err) + } + gitDir, err := rfs.RepoPath(user.Username, repo.Name) + if err != nil { + t.Fatalf("RepoPath: %v", err) + } + if err := initBareRepo(gitDir); err != nil { + t.Fatalf("init bare: %v", err) + } + commitOID, err := repogit.InitialCommit{ + GitDir: gitDir, + AuthorName: "Test Author", + AuthorEmail: "test@example.com", + Message: "seed index fixture", + Branch: "trunk", + When: time.Date(2026, 5, 28, 12, 0, 0, 0, time.UTC), + Files: []repogit.FileEntry{ + {Path: "small.txt", Body: []byte("searchable needle\n")}, + {Path: "huge.txt", Body: []byte(strings.Repeat("x", 256*1024+1))}, + }, + }.Build(ctx) + if err != nil { + t.Fatalf("InitialCommit.Build: %v", err) + } + + handler := jobs.RepoIndexCode(jobs.IndexCodeDeps{ + Pool: pool, + RepoFS: rfs, + Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), + }) + payload, _ := json.Marshal(jobs.IndexCodePayload{RepoID: repo.ID}) + if err := handler(ctx, payload); err != nil { + t.Fatalf("repo:index_code: %v", err) + } + + var pathCount int + if err := pool.QueryRow(ctx, ` + SELECT count(*) FROM code_search_paths + WHERE repo_id = $1 AND path IN ('small.txt', 'huge.txt') + `, repo.ID).Scan(&pathCount); err != nil { + t.Fatalf("count paths: %v", err) + } + if pathCount != 2 { + t.Fatalf("path count = %d, want 2", pathCount) + } + + var contentCount int + if err := pool.QueryRow(ctx, ` + SELECT count(*) FROM code_search_content + WHERE repo_id = $1 + `, repo.ID).Scan(&contentCount); err != nil { + t.Fatalf("count content: %v", err) + } + if contentCount != 1 { + t.Fatalf("content count = %d, want 1", contentCount) + } + + indexedRepo, err := rq.GetRepoByID(ctx, pool, repo.ID) + if err != nil { + t.Fatalf("GetRepoByID: %v", err) + } + if !indexedRepo.LastIndexedOid.Valid || indexedRepo.LastIndexedOid.String != commitOID { + t.Fatalf("last_indexed_oid = %#v, want %q", indexedRepo.LastIndexedOid, commitOID) + } +}