From 94e91422991d24c149b2f3d1f8e5dfd423c672cb Mon Sep 17 00:00:00 2001 From: Ayaan Zaidi Date: Thu, 17 Sep 2026 18:43:27 +0530 Subject: [PATCH] feat(portable): consume backed-up sources during export --- docs/portable-stores.md | 17 +- internal/cli/help.go | 2 +- internal/cli/portable_commands.go | 2 + internal/cli/portable_consume_unix_test.go | 57 ++++++ internal/portable/consume_unix.go | 75 ++++++++ internal/portable/consume_unix_test.go | 194 +++++++++++++++++++++ internal/portable/consume_windows.go | 10 ++ internal/portable/export.go | 13 +- 8 files changed, 366 insertions(+), 4 deletions(-) create mode 100644 internal/cli/portable_consume_unix_test.go create mode 100644 internal/portable/consume_unix.go create mode 100644 internal/portable/consume_unix_test.go create mode 100644 internal/portable/consume_windows.go diff --git a/docs/portable-stores.md b/docs/portable-stores.md index cf747c7..8ee8198 100644 --- a/docs/portable-stores.md +++ b/docs/portable-stores.md @@ -299,7 +299,7 @@ After pruning, commit and push both the database and its `.manifest.json` from t ## Derived generations: `gitcrawl portable export` `portable export` creates a new, validated database-and-manifest generation from -the configured active database without changing that database. It is generic +the configured active database, preserving the source by default. It is generic artifact production: Gitcrawl owns the consistent SQLite snapshot, semantic shaping, validation, size budget, digest, and manifest. Promotion into a repository, replacement of an older generation, Git commits, and publication @@ -313,6 +313,20 @@ deletion because it is never exposed and is deleted on any error. Privacy and durability come from the separate compact generation, full validation, hashing, fsync, and atomic directory commit. +For a disposable batch-job database with a verified external recovery copy, +`--consume-source` moves the source into export staging instead of creating the +initial full-size copy. Close all database users, checkpoint SQLite, and set +`PRAGMA journal_mode=DELETE` before saving and verifying the recovery copy. +The source must be a regular file with one hard link, no SQLite sidecars, and +on the same filesystem as the output. An exclusive SQLite lock rejects active +readers and writers. This mode is supported on Unix systems. + +Once the source moves, its original path is gone. Export errors discard the +consumed staging file; recover from the verified backup. Do not save the pruned +file over the full-runtime backup. All shaping, privacy checks, compaction, +validation, and artifact publication remain the same as ordinary export. +Compaction still needs space for the reduced database and compressed artifact. + ```bash gitcrawl --config /path/to/config.toml portable export \ --profile current-state-v1 \ @@ -448,6 +462,7 @@ the private staging directory. | `--repository ` | _(unset)_ | Semantically restrict the artifact to exactly one repository | | `--body-chars ` | `256` | Maximum body characters retained in compact excerpts | | `--max-bytes ` | _(unset)_ | Inclusive maximum finalized database size | +| `--consume-source` | _(off)_ | Move an exclusively owned, backed-up source into staging; errors after handoff discard it | | `--json` | _(off)_ | Stable structured result, including local source and output paths | ## A typical publishing flow diff --git a/internal/cli/help.go b/internal/cli/help.go index 6e2841a..b84a1ca 100644 --- a/internal/cli/help.go +++ b/internal/cli/help.go @@ -310,7 +310,7 @@ const portableUsageText = `gitcrawl portable manages local portable-store snapsh Usage: gitcrawl portable refresh --expected-remote URL [--store-dir PATH] [--portable-db PATH] [--branch main] [--git PATH] [--timeout 2m] [--min-free-bytes N] [--max-growth-bytes N] [--json] gitcrawl portable prune [--body-chars N] [--no-vacuum] [--include-sync-failures] [--no-publish] [--json] - gitcrawl portable export --profile current-state-v1 --output-dir PATH [--repository owner/repo] [--database-name NAME] [--public-path PATH] [--body-chars N] [--max-bytes N] [--compression gzip] [--max-archive-bytes N] [--json] + gitcrawl portable export --profile current-state-v1 --output-dir PATH [--repository owner/repo] [--database-name NAME] [--public-path PATH] [--body-chars N] [--max-bytes N] [--compression gzip] [--max-archive-bytes N] [--consume-source] [--json] Subcommands: refresh validate and fast-forward a clean configured subscriber diff --git a/internal/cli/portable_commands.go b/internal/cli/portable_commands.go index 8c084bf..45c6f96 100644 --- a/internal/cli/portable_commands.go +++ b/internal/cli/portable_commands.go @@ -46,6 +46,7 @@ func (a *App) runPortableExport(ctx context.Context, args []string) error { maxBytesRaw := fs.String("max-bytes", "", "maximum finalized database bytes") compression := fs.String("compression", "", "artifact compression (gzip)") maxArchiveBytesRaw := fs.String("max-archive-bytes", "", "maximum finalized archive bytes") + consumeSource := fs.Bool("consume-source", false, "consume an exclusively owned, backed-up database instead of copying it") jsonOut := fs.Bool("json", false, "write JSON output") valueFlags := map[string]bool{ "profile": true, "body-chars": true, "output-dir": true, @@ -139,6 +140,7 @@ func (a *App) runPortableExport(ctx context.Context, args []string) error { MaxBytes: maxBytes, Compression: *compression, MaxArchiveBytes: maxArchiveBytes, + ConsumeSource: *consumeSource, Progress: func(stage portableexport.Stage) { now := time.Now() fmt.Fprintf( diff --git a/internal/cli/portable_consume_unix_test.go b/internal/cli/portable_consume_unix_test.go new file mode 100644 index 0000000..4898d3c --- /dev/null +++ b/internal/cli/portable_consume_unix_test.go @@ -0,0 +1,57 @@ +//go:build !windows + +package cli + +import ( + "bytes" + "context" + "database/sql" + "encoding/json" + "errors" + "os" + "path/filepath" + "testing" +) + +func TestPortableExportCommandConsumesClosedSource(t *testing.T) { + ctx := context.Background() + dir := t.TempDir() + config := filepath.Join(dir, "config.toml") + source := filepath.Join(dir, "source.db") + app := New() + var stdout, stderr bytes.Buffer + app.Stdout, app.Stderr = &stdout, &stderr + if err := app.Run(ctx, []string{"--config", config, "init", "--db", source}); err != nil { + t.Fatal(err) + } + seedPortableThread(t, source, 7, "consumed source") + db, err := sql.Open("sqlite", source) + if err != nil { + t.Fatal(err) + } + if _, err := db.Exec(`pragma journal_mode=delete`); err != nil { + t.Fatal(err) + } + if err := db.Close(); err != nil { + t.Fatal(err) + } + stdout.Reset() + if err := app.Run(ctx, []string{"--config", config, "portable", "export", "--profile", "current-state-v1", + "--output-dir", filepath.Join(dir, "artifact"), "--consume-source", "--json"}); err != nil { + t.Fatal(err) + } + var result struct { + SourceConsumed bool `json:"source_consumed"` + ArtifactCommitted bool `json:"artifact_committed"` + IntegrityCheck string `json:"integrity_check"` + } + if err := json.Unmarshal(stdout.Bytes(), &result); err != nil { + t.Fatal(err) + } + if !result.SourceConsumed || !result.ArtifactCommitted || result.IntegrityCheck != "ok" { + t.Fatalf("result: %+v", result) + } + if _, err := os.Stat(source); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("source still exists: %v", err) + } +} diff --git a/internal/portable/consume_unix.go b/internal/portable/consume_unix.go new file mode 100644 index 0000000..4ae7d80 --- /dev/null +++ b/internal/portable/consume_unix.go @@ -0,0 +1,75 @@ +//go:build !windows + +package portable + +import ( + "context" + "database/sql" + "errors" + "fmt" + "net/url" + "os" + "path/filepath" + "syscall" +) + +// The caller owns the source exclusively and has already saved its recovery +// copy. After the rename, every failure discards the consumed staging file. +func consumeSQLite(ctx context.Context, sourcePath, targetPath string) error { + source, err := os.Lstat(sourcePath) + if err != nil { + return fmt.Errorf("inspect consumed source: %w", err) + } + if !source.Mode().IsRegular() || source.Sys().(*syscall.Stat_t).Nlink != 1 { + return fmt.Errorf("consume-source requires a regular file with exactly one link") + } + parent, err := os.Stat(filepath.Dir(targetPath)) + if err != nil { + return fmt.Errorf("inspect consume-source staging filesystem: %w", err) + } + if source.Sys().(*syscall.Stat_t).Dev != parent.Sys().(*syscall.Stat_t).Dev { + return fmt.Errorf("consume-source requires source and output on the same filesystem") + } + for _, suffix := range []string{"-wal", "-shm", "-journal"} { + if _, err := os.Lstat(sourcePath + suffix); !errors.Is(err, os.ErrNotExist) { + return fmt.Errorf("consume-source requires a closed database without SQLite sidecars") + } + } + u := url.URL{Scheme: "file", Path: sourcePath} + query := u.Query() + query.Set("mode", "rw") + query.Add("_pragma", "busy_timeout(0)") + u.RawQuery = query.Encode() + db, err := sql.Open("sqlite", u.String()) + if err != nil { + return fmt.Errorf("open consumed source: %w", err) + } + defer db.Close() + db.SetMaxOpenConns(1) + var mode string + if err := db.QueryRowContext(ctx, `pragma journal_mode`).Scan(&mode); err != nil { + return fmt.Errorf("read consumed source journal mode: %w", err) + } + if mode != "delete" { + return fmt.Errorf("consume-source requires a checkpointed database in journal_mode=delete") + } + // A rollback-mode exclusive transaction rejects active readers and writers. + // It makes no writes and holds the SQLite file lock through the rename. + if _, err := db.ExecContext(ctx, `begin exclusive`); err != nil { + return fmt.Errorf("lock consumed source exclusively: %w", err) + } + defer db.ExecContext(context.Background(), `rollback`) + current, err := os.Lstat(sourcePath) + if err != nil || !os.SameFile(source, current) || current.Sys().(*syscall.Stat_t).Nlink != 1 || + source.Size() != current.Size() || !source.ModTime().Equal(current.ModTime()) { + return fmt.Errorf("consume-source file changed before handoff") + } + if err := ctx.Err(); err != nil { + return err + } + // Rename is the ownership transfer. In particular, EXDEV must not copy. + if err := os.Rename(sourcePath, targetPath); err != nil { + return fmt.Errorf("move consumed source into export staging: %w", err) + } + return nil +} diff --git a/internal/portable/consume_unix_test.go b/internal/portable/consume_unix_test.go new file mode 100644 index 0000000..5d41273 --- /dev/null +++ b/internal/portable/consume_unix_test.go @@ -0,0 +1,194 @@ +//go:build !windows + +package portable + +import ( + "context" + "errors" + "os" + "path/filepath" + "strings" + "testing" +) + +func consumeFixture(t *testing.T) ExportOptions { + t.Helper() + dir := t.TempDir() + source := filepath.Join(dir, "runtime.db") + st := seedExportSource(t, context.Background(), source) + if err := st.Close(); err != nil { + t.Fatal(err) + } + db := openRawDB(t, source) + if _, err := db.Exec(`pragma journal_mode=delete`); err != nil { + t.Fatal(err) + } + if err := db.Close(); err != nil { + t.Fatal(err) + } + return ExportOptions{SourceDBPath: source, OutputDir: filepath.Join(dir, "consumed"), + DatabaseName: "archive.db", PublicPath: "data/archive.db", Profile: CurrentStateV1, + Repository: "openclaw/gitcrawl", BodyChars: 8, Compression: CompressionGzip, ConsumeSource: true} +} + +func TestConsumingExportMatchesSourcePreservingExport(t *testing.T) { + options := consumeFixture(t) + preserved := options + preserved.ConsumeSource = false + preserved.OutputDir += "-preserved" + before, err := os.Stat(options.SourceDBPath) + if err != nil { + t.Fatal(err) + } + want, err := Export(context.Background(), preserved) + if err != nil { + t.Fatal(err) + } + if want.SourceConsumed { + t.Fatal("ordinary export consumed its source") + } + var transferred bool + options.Progress = func(stage Stage) { + if stage != StageRepositoryScope { + return + } + if _, err := os.Stat(options.SourceDBPath); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("source still exists after handoff: %v", err) + } + stages, err := filepath.Glob(filepath.Join(filepath.Dir(options.OutputDir), ".gitcrawl-portable-export-*", options.DatabaseName)) + if err != nil || len(stages) != 1 { + t.Fatalf("private stage: %v %v", stages, err) + } + after, err := os.Stat(stages[0]) + if err != nil || !os.SameFile(before, after) { + t.Fatalf("export copied rather than moved the source: %v", err) + } + transferred = true + } + got, err := Export(context.Background(), options) + if err != nil { + t.Fatal(err) + } + if !transferred || !got.SourceConsumed || !got.ArtifactCommitted || got.ArtifactID != want.ArtifactID || + got.QuickCheck != "ok" || got.IntegrityCheck != "ok" || got.ForeignKeyViolations != 0 { + t.Fatalf("consuming export differs: got %+v; want identity %s", got, want.ArtifactID) + } +} + +func TestConsumingExportRejectsUnsafeSourceBeforeHandoff(t *testing.T) { + for _, scenario := range []string{"hardlink", "symlink", "wal", "shm", "journal", "writer", "reader", "wal-mode", "cancelled"} { + t.Run(scenario, func(t *testing.T) { + options := consumeFixture(t) + source := options.SourceDBPath + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + switch scenario { + case "hardlink": + if err := os.Link(source, source+".alias"); err != nil { + t.Fatal(err) + } + case "symlink": + if err := os.Symlink(source, source+".alias"); err != nil { + t.Fatal(err) + } + options.SourceDBPath += ".alias" + case "wal", "shm", "journal": + if err := os.WriteFile(source+"-"+scenario, []byte("pending"), 0o600); err != nil { + t.Fatal(err) + } + case "writer", "reader": + db := openRawDB(t, source) + defer db.Close() + statement := `begin immediate` + if scenario == "reader" { + statement = `begin; select count(*) from threads` + } + if _, err := db.Exec(statement); err != nil { + t.Fatal(err) + } + defer db.Exec(`rollback`) + case "wal-mode": + db := openRawDB(t, source) + if _, err := db.Exec(`pragma journal_mode=wal`); err != nil { + t.Fatal(err) + } + if err := db.Close(); err != nil { + t.Fatal(err) + } + case "cancelled": + cancel() + } + before := readFile(t, source) + result, err := Export(ctx, options) + if err == nil || result.SourceConsumed || result.ArtifactCommitted { + t.Fatalf("unsafe consume accepted: %+v %v", result, err) + } + if string(before) != string(readFile(t, source)) { + t.Fatal("rejected export changed source bytes") + } + if _, err := os.Stat(options.OutputDir); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("output exists: %v", err) + } + }) + } +} + +func TestConsumingExportFailureDiscardsOnlyConsumedStage(t *testing.T) { + for _, stage := range []Stage{StageRepositoryScope, StageCanonicalShaping, StageFinalVacuum, StageManifest} { + t.Run(string(stage), func(t *testing.T) { + options := consumeFixture(t) + checkpoint := readFile(t, options.SourceDBPath) + backup := options.SourceDBPath + ".checkpoint" + if err := os.WriteFile(backup, checkpoint, 0o600); err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + options.Progress = func(current Stage) { + if current == stage { + cancel() + } + } + result, err := Export(ctx, options) + if !errors.Is(err, context.Canceled) || !result.SourceConsumed || result.ArtifactCommitted { + t.Fatalf("failure result: %+v %v", result, err) + } + if _, err := os.Stat(options.SourceDBPath); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("source restored incorrectly: %v", err) + } + stages, err := filepath.Glob(filepath.Join(filepath.Dir(options.OutputDir), ".gitcrawl-portable-export-*")) + if err != nil || len(stages) != 0 { + t.Fatalf("stage leaked: %v %v", stages, err) + } + if string(checkpoint) != string(readFile(t, backup)) { + t.Fatal("checkpoint changed") + } + if err := os.Rename(backup, options.SourceDBPath); err != nil { + t.Fatal(err) + } + options.Progress = nil + if _, err := Export(context.Background(), options); err != nil { + t.Fatalf("recovery export: %v", err) + } + }) + } +} + +func TestConsumeSourceRejectsDifferentFilesystem(t *testing.T) { + if info, err := os.Stat("/dev/shm"); err != nil || !info.IsDir() { + t.Skip("requires a separate shared-memory filesystem") + } + options := consumeFixture(t) + dir, err := os.MkdirTemp("/dev/shm", "gitcrawl-consume-test-") + if err != nil { + t.Skipf("shared-memory filesystem unavailable: %v", err) + } + defer os.RemoveAll(dir) + err = consumeSQLite(context.Background(), options.SourceDBPath, filepath.Join(dir, "artifact.db")) + if err == nil || !strings.Contains(err.Error(), "same filesystem") { + t.Fatalf("cross-filesystem handoff: %v", err) + } + if _, err := os.Stat(options.SourceDBPath); err != nil { + t.Fatalf("source lost: %v", err) + } +} diff --git a/internal/portable/consume_windows.go b/internal/portable/consume_windows.go new file mode 100644 index 0000000..40a6d1a --- /dev/null +++ b/internal/portable/consume_windows.go @@ -0,0 +1,10 @@ +package portable + +import ( + "context" + "fmt" +) + +func consumeSQLite(_ context.Context, _, _ string) error { + return fmt.Errorf("consume-source is not supported on Windows") +} diff --git a/internal/portable/export.go b/internal/portable/export.go index 9a06496..e461d0a 100644 --- a/internal/portable/export.go +++ b/internal/portable/export.go @@ -36,6 +36,7 @@ type ExportOptions struct { MaxBytes *int64 Compression string MaxArchiveBytes *int64 + ConsumeSource bool Progress ProgressFunc } @@ -63,6 +64,7 @@ type ExportResult struct { PortableSchema string `json:"portable_schema"` Schema string `json:"schema"` SourceDBPath string `json:"source_db_path"` + SourceConsumed bool `json:"source_consumed,omitempty"` OutputDir string `json:"output_dir"` DatabasePath string `json:"database_path"` ManifestPath string `json:"manifest_path"` @@ -192,8 +194,15 @@ func (e exporter) export(ctx context.Context, options ExportOptions) (result Exp if err := reportProgress(ctx, options.Progress, StageSnapshot); err != nil { return result, err } - if err := snapshotSQLite(ctx, sourcePath, dbPath); err != nil { - return result, err + if options.ConsumeSource { + if err := consumeSQLite(ctx, sourcePath, dbPath); err != nil { + return result, err + } + result.SourceConsumed = true + } else { + if err := snapshotSQLite(ctx, sourcePath, dbPath); err != nil { + return result, err + } } // Opening the disposable snapshot migrates valid older/physically-pruned // portable schemas back to the current writable schema before shaping.