feat(portable): consume backed-up sources during export

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 <owner/repo>` | _(unset)_ | Semantically restrict the artifact to exactly one repository |
 | `--body-chars <n>` | `256` | Maximum body characters retained in compact excerpts |
 | `--max-bytes <n>` | _(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

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

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(

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)
+	}
+}

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
+}

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)
+	}
+}