diff --git a/CHANGELOG.md b/CHANGELOG.md index 5855131f..9e9d4ad1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,10 @@ ## Unreleased +- Request `synchronous=FULL` on every SQLite connection and, on macOS, `fullfsync` for commits and checkpoints. Go test binaries skip the macOS full flush. +- Let the native code producer publish from a captured upstream commit when core progress is fresh, even while provider coverage is incomplete. Retries and coverage are not changed. +- Add `portable export --compression split`, which publishes the uncompressed portable database as verified 32 MiB parts so Git stores each new generation as a small delta instead of a new 68 MB gzip blob. Refresh, init, status and recovery read split stores; older readers refuse them and keep their last good mirror. After advancing to a split generation, strict refresh geometrically repacks recent objects so subscribers also keep each generation as a delta, skipping a loose backlog of more than 512 objects. Raw and gzip stores are still never repacked. + ## 0.16.0 - 2026-10-20 **Highlights:** Time-filtered neighbor discovery, Go security fixes, and reliable large-result lookups. diff --git a/docs/portable-stores.md b/docs/portable-stores.md index 8ee81980..e4b5c75c 100644 --- a/docs/portable-stores.md +++ b/docs/portable-stores.md @@ -55,6 +55,24 @@ size and SHA-256, runs SQLite `quick_check`, and only then atomically replaces the runtime mirror. Legacy raw SQLite stores continue to use the same manifest without compression fields. +Stores may instead commit a split artifact: `compression: "split"` and an +ordered `archiveParts` list of `{path, bytes, sha256}` basenames beside the +manifest (`NAME.part-000`, `NAME.part-001`, ...). The parts are the +uncompressed SQLite bytes, so `archiveBytes`/`archiveSha256` equal +`outputBytes`/`sha256`. Gitcrawl checks every part's size and digest, the digest +of their ordered concatenation and the per-part `maxArchiveBytes`, then +reassembles and validates the database like any other artifact. Readers older +than split support refuse the manifest as an unsupported compression and keep +their last good runtime mirror. + +Split exists for Git-hosted stores. Each gzip generation is a new compressed +blob that cannot delta against its predecessor, so every hourly snapshot added +about 68 MB to the store. Consecutive uncompressed parts share most bytes; a +measured pair of consecutive production generations transferred as a 0.5 MB +thin pack and added about 0.5 MB to a fully packed repository. Subscribers +receive each new part as a full object; strict refresh's post-advance geometric +repack then folds it into deltas (see below). + Initialization validates portable arguments before invoking Git and validates the artifact before saving configuration. Repeated initialization and a publisher's raw-to-gzip transition do not require a raw `.db` in the checkout. @@ -106,8 +124,10 @@ checkout files**, hidden index entries, unrelated origins, divergent history, active Git locks or operation state, orphan/temporary packs, hooks, filters, attributes, submodules, linked worktrees, alternate object stores, config includes/redirections and unsupported extensions. Resolve these deliberately -outside the subscriber command. It never resets, cleans, prunes, repacks, -reclones, deletes backups or sidecars, or invokes reader/doctor auto-repair. +outside the subscriber command. It never resets, cleans, prunes, reclones, +deletes backups or sidecars, or invokes reader/doctor auto-repair. The only +repack is the bounded post-advance geometric repack of a split store described +below; raw and gzip stores are never repacked. Admission examines every exposed Git configuration scope. Refusals identify the scope and unsafe category without printing keys or values. Single-valued @@ -185,14 +205,28 @@ operator inspection before another strict refresh. JSON reports `stage`, `result` (`updated`, `no-op`, `refused`, `partial`), `before_commit`, `after_commit`, `target_commit`, `artifact_id`, `sha256`, -`artifact_bytes`, `mirror_destination`, `mirror_result`, `capacity` and -`elapsed_ms`. A refusal includes a bounded `reason` and exits nonzero. Stderr +`artifact_bytes`, `mirror_destination`, `mirror_result`, `capacity`, +`elapsed_ms` and, for split stores, `repack`. A refusal includes a bounded `reason` and exits nonzero. Stderr contains stage/elapsed/result diagnostics; stdout remains data. A no-op means the refs already match; a missing runtime may still be materialized. Git failures use fixed, credential-safe repository, authentication, connection or disk-space diagnostics when recognized, with general troubleshooting guidance otherwise; raw remote/helper stderr is never included. +When the checkout's generation is split, strict refresh runs +`git repack -d --geometric=2` inside the same deadline and growth budget, so +new generations are stored as deltas against their predecessors +(`repack: "geometric"`). This also covers objects fetched by an implicit reader +fast-forward, which never repacks, and generations kept as fetched packs +(for example with `fetch.unpackLimit`), so a no-op refresh can still +consolidate. With nothing loose and a single pack it reports +`repack: "current"` and does nothing. A larger +history pack is left untouched. With more than 512 loose objects, such as years of loose gzip +generations, refresh skips this (`skipped-loose-backlog`); pack that backlog +once with an owner-run `git repack -d` first. A consolidation failure is reported +as `repack: "failed"` with a bounded `repack_reason` and next step, without +undoing the validated generation. + All portable Git entry points explicitly set `maintenance.auto=false` and `gc.auto=0`, including init, implicit reader fetch/merge and retained recovery clone/reset helpers. Fetch also uses `--no-auto-maintenance`. Hooks, fsmonitor @@ -350,7 +384,10 @@ host path. `--body-chars` defaults to `256`. `--max-bytes` is an optional, inclusive limit for raw database generations. `--compression gzip` instead commits only the gzip archive and manifest in the generation. Pair it with `--max-archive-bytes`; a Git host requiring an artifact smaller than -100,000,000 bytes should pass `99999999`. +100,000,000 bytes should pass `99999999`. `--compression split` commits the +manifest and ordered uncompressed parts of 32 MiB (or `--max-archive-bytes`, +when smaller); `--max-archive-bytes` then caps each part, and more than 64 parts +is refused. The optional `--repository owner/repo` is semantic export behavior: Gitcrawl removes all other repositories and their dependent rows from the disposable snapshot, verifies the exact remaining identity, and records it in the manifest. @@ -372,7 +409,9 @@ file SHA while retaining the same meaningful portable state. For gzip generations, the manifest also records `compression`, `archivePath`, `archiveBytes`, `archiveSha256`, and `maxArchiveBytes`. Gitcrawl validates the archive against both compressed and expanded identities before atomically -committing the generation. +committing the generation. Split generations record `archiveParts` instead of +`archivePath`; JSON command output lists them as `archive_parts` with output +paths. Destructive `portable prune` continues to record its operation time as `portable_metadata.exported_at`. diff --git a/internal/cli/help.go b/internal/cli/help.go index 9fe75c02..ff76fad0 100644 --- a/internal/cli/help.go +++ b/internal/cli/help.go @@ -345,7 +345,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] [--consume-source] [--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|split] [--max-archive-bytes N] [--consume-source] [--json] Subcommands: refresh validate and fast-forward a clean configured subscriber @@ -361,6 +361,10 @@ Export never changes the active database or publishes the artifact. It creates a complete database and manifest generation at a previously nonexistent path. With --compression gzip, it commits the gzip archive and manifest without the uncompressed database; --max-archive-bytes applies to that published archive. +With --compression split, it commits the uncompressed database as ordered +32 MiB parts (NAME.part-000, ...) and the manifest. Git can store a later +generation as a small delta against these parts; --max-archive-bytes caps each +part. Readers reassemble and verify every part and the whole database. --repository semantically restricts the disposable snapshot to one owner/repo. Refresh never regenerates config or invokes repair. It requires a matching origin, clean checkout and manifest-backed artifact. --git requires an absolute diff --git a/internal/cli/portable_commands.go b/internal/cli/portable_commands.go index 45c6f96c..49582666 100644 --- a/internal/cli/portable_commands.go +++ b/internal/cli/portable_commands.go @@ -44,7 +44,7 @@ func (a *App) runPortableExport(ctx context.Context, args []string) error { publicPath := fs.String("public-path", "", "logical portable database path") repositoryRaw := fs.String("repository", "", "restrict the artifact to one owner/repo") maxBytesRaw := fs.String("max-bytes", "", "maximum finalized database bytes") - compression := fs.String("compression", "", "artifact compression (gzip)") + compression := fs.String("compression", "", "artifact packaging (gzip or split)") 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") @@ -89,8 +89,8 @@ func (a *App) runPortableExport(ctx context.Context, args []string) error { } maxArchiveBytes = &parsed } - if *compression != "" && *compression != portableexport.CompressionGzip { - return usageErr(fmt.Errorf("--compression must be %q", portableexport.CompressionGzip)) + if *compression != "" && *compression != portableexport.CompressionGzip && *compression != portableexport.CompressionSplit { + return usageErr(fmt.Errorf("--compression must be %q or %q", portableexport.CompressionGzip, portableexport.CompressionSplit)) } if maxArchiveBytes != nil && *compression == "" { return usageErr(fmt.Errorf("--max-archive-bytes requires --compression")) diff --git a/internal/cli/portable_manifest.go b/internal/cli/portable_manifest.go index 996e9e83..2076290b 100644 --- a/internal/cli/portable_manifest.go +++ b/internal/cli/portable_manifest.go @@ -2,9 +2,11 @@ package cli import ( "context" + "crypto/sha256" "encoding/json" "errors" "fmt" + "io" "os" "path/filepath" "strings" @@ -26,6 +28,73 @@ type portableDBManifest struct { ArchivePath string `json:"archivePath,omitempty"` ArchiveBytes int64 `json:"archiveBytes,omitempty"` ArchiveSHA256 string `json:"archiveSha256,omitempty"` + // Split archives list ordered uncompressed parts beside the manifest. + ArchiveParts []portableArchivePart `json:"archiveParts,omitempty"` + MaxArchiveBytes int64 `json:"maxArchiveBytes,omitempty"` +} + +type portableArchivePart struct { + Path string `json:"path"` + Bytes int64 `json:"bytes"` + SHA256 string `json:"sha256"` +} + +const ( + portableMaxArchiveParts = 64 + // Each committed part stays below the hosting per-file limit. + portableMaxArchivePartBytes = 99999999 +) + +// portableArchivePartPaths returns the ordered part paths of a split archive +// whose first part is firstPart (as returned by portableSourceArtifact). +func portableArchivePartPaths(firstPart string, manifest portableDBManifest) []string { + paths := make([]string, 0, len(manifest.ArchiveParts)) + for _, part := range manifest.ArchiveParts { + paths = append(paths, filepath.Join(filepath.Dir(firstPart), part.Path)) + } + return paths +} + +func portableSplitArchive(manifest portableDBManifest) bool { + return strings.TrimSpace(manifest.Compression) == portableexport.CompressionSplit +} + +func validatePortableSplitMetadata(manifest portableDBManifest) error { + if strings.TrimSpace(manifest.ArchivePath) != "" { + return fmt.Errorf("portable manifest mismatch: split archive must not name archivePath") + } + if len(manifest.ArchiveParts) == 0 || len(manifest.ArchiveParts) > portableMaxArchiveParts { + return fmt.Errorf("portable manifest mismatch: split archive must have 1..%d parts", portableMaxArchiveParts) + } + if manifest.OutputBytes <= 0 || manifest.ArchiveBytes != manifest.OutputBytes || + !strings.EqualFold(strings.TrimSpace(manifest.ArchiveSHA256), strings.TrimSpace(manifest.SHA256)) || + strings.TrimSpace(manifest.SHA256) == "" { + return fmt.Errorf("portable manifest mismatch: split archive must equal the database size and sha256") + } + var total int64 + seen := map[string]bool{} + for index, part := range manifest.ArchiveParts { + name := part.Path + if name == "" || name == "." || name == ".." || strings.ContainsAny(name, "/\\\x00:") || seen[name] { + return fmt.Errorf("portable manifest mismatch: archive part %d path must be a unique basename", index) + } + seen[name] = true + if part.Bytes <= 0 || part.Bytes > portableMaxArchivePartBytes || + manifest.MaxArchiveBytes > 0 && part.Bytes > manifest.MaxArchiveBytes { + return fmt.Errorf("portable manifest mismatch: archive part %d size is outside its budget", index) + } + if len(strings.TrimSpace(part.SHA256)) != 64 { + return fmt.Errorf("portable manifest mismatch: archive part %d sha256 missing", index) + } + if total > manifest.OutputBytes-part.Bytes { + return fmt.Errorf("portable manifest mismatch: archive parts exceed outputBytes") + } + total += part.Bytes + } + if total != manifest.OutputBytes { + return fmt.Errorf("portable manifest mismatch: archive parts total %d != %d", total, manifest.OutputBytes) + } + return nil } func portableDBManifestPath(dbPath string) string { @@ -153,8 +222,21 @@ func portableSourceArtifact(dbPath string) (string, portableDBManifest, bool, er return "", portableDBManifest{}, false, fmt.Errorf("portable manifest mismatch: %w", err) } if !ok || strings.TrimSpace(manifest.Compression) == "" { + if ok && len(manifest.ArchiveParts) > 0 { + return "", portableDBManifest{}, false, fmt.Errorf("portable manifest mismatch: archiveParts require split compression") + } return dbPath, manifest, false, nil } + if portableSplitArchive(manifest) { + if err := validatePortableSplitMetadata(manifest); err != nil { + return "", portableDBManifest{}, false, err + } + // Callers receive the first part; portableArchivePartPaths lists all. + return filepath.Join(filepath.Dir(manifestPath), manifest.ArchiveParts[0].Path), manifest, true, nil + } + if len(manifest.ArchiveParts) > 0 { + return "", portableDBManifest{}, false, fmt.Errorf("portable manifest mismatch: archiveParts require split compression") + } if strings.TrimSpace(manifest.Compression) != "gzip" { return "", portableDBManifest{}, false, fmt.Errorf( "portable manifest mismatch: unsupported compression %q", @@ -200,6 +282,9 @@ func validatePortableArchive(path string, manifest portableDBManifest) error { } func validatePortableArchiveContext(ctx context.Context, path string, manifest portableDBManifest) error { + if portableSplitArchive(manifest) { + return validatePortableSplitArchiveContext(ctx, path, manifest) + } info, err := os.Stat(path) if err != nil { return err @@ -225,3 +310,39 @@ func validatePortableArchiveContext(ctx context.Context, path string, manifest p } return nil } + +// validatePortableSplitArchiveContext checks every part's size and digest and +// the digest of their ordered concatenation. +func validatePortableSplitArchiveContext(ctx context.Context, firstPart string, manifest portableDBManifest) error { + if err := validatePortableSplitMetadata(manifest); err != nil { + return err + } + whole := sha256.New() + for index, path := range portableArchivePartPaths(firstPart, manifest) { + part := manifest.ArchiveParts[index] + info, err := os.Stat(path) + if err != nil { + return err + } + if !info.Mode().IsRegular() || info.Size() != part.Bytes { + return fmt.Errorf("portable manifest mismatch: archive part %d size %d != %d", index, info.Size(), part.Bytes) + } + file, err := os.Open(path) + if err != nil { + return err + } + hash := sha256.New() + written, copyErr := io.Copy(io.MultiWriter(hash, whole), portableContextReader{ctx: ctx, reader: io.LimitReader(file, part.Bytes+1)}) + closeErr := file.Close() + if err := errors.Join(copyErr, closeErr); err != nil { + return err + } + if written != part.Bytes || !strings.EqualFold(fmt.Sprintf("%x", hash.Sum(nil)), strings.TrimSpace(part.SHA256)) { + return fmt.Errorf("portable manifest mismatch: archive part %d sha256 mismatch", index) + } + } + if sum := fmt.Sprintf("%x", whole.Sum(nil)); !strings.EqualFold(sum, strings.TrimSpace(manifest.ArchiveSHA256)) { + return fmt.Errorf("portable manifest mismatch: archive sha256 %s != %s", sum, manifest.ArchiveSHA256) + } + return nil +} diff --git a/internal/cli/portable_recovery.go b/internal/cli/portable_recovery.go index e2eeb0dc..4cd348cd 100644 --- a/internal/cli/portable_recovery.go +++ b/internal/cli/portable_recovery.go @@ -173,8 +173,12 @@ func preserveMalformedPortableDB(root, dbPath string) (string, error) { dbPath + "-shm", dbPath + ".manifest.json", } - if archivePath, _, compressed, err := portableSourceArtifact(dbPath); err == nil && compressed { - paths = append(paths, archivePath) + if archivePath, manifest, compressed, err := portableSourceArtifact(dbPath); err == nil && compressed { + if portableSplitArchive(manifest) { + paths = append(paths, portableArchivePartPaths(archivePath, manifest)...) + } else { + paths = append(paths, archivePath) + } } for _, path := range paths { if _, err := os.Stat(path); err != nil { diff --git a/internal/cli/portable_refresh.go b/internal/cli/portable_refresh.go index 3d636d54..f500b5b2 100644 --- a/internal/cli/portable_refresh.go +++ b/internal/cli/portable_refresh.go @@ -8,6 +8,7 @@ import ( "io" "os" "path/filepath" + "strconv" "strings" "time" @@ -39,6 +40,55 @@ type portableRefreshResult struct { Capacity portableCapacity `json:"capacity"` ElapsedMS int64 `json:"elapsed_ms"` Reason string `json:"reason,omitempty"` + // Repack reports split-store object consolidation: geometric, current + // (no loose objects), skipped-loose-backlog, or failed. It never affects + // the validated generation. + Repack string `json:"repack,omitempty"` + // RepackReason is a bounded, credential-safe cause for repack=failed. + RepackReason string `json:"repack_reason,omitempty"` +} + +// portableRepackLooseLimit bounds the loose objects one refresh will pack. +// A larger backlog (for example years of loose gzip generations) needs a +// separate owner-run repack instead of a single bounded refresh. +const portableRepackLooseLimit = 512 + +// consolidatePortableObjects lets Git store a new split generation as deltas +// against its predecessor. Fetched objects arrive loose or in packs whose +// bases are full copies; a geometric repack folds recent packs together with +// delta search while leaving a much larger history pack untouched. +func consolidatePortableObjects(ctx context.Context, root string) (string, error) { + output, err := portableGitOutput(ctx, root, "count-objects", "-v") + if err != nil { + return "failed", err + } + loose, packs := int64(-1), int64(-1) + for _, line := range strings.Split(output, "\n") { + for prefix, target := range map[string]*int64{"count: ": &loose, "packs: ": &packs} { + if value, ok := strings.CutPrefix(line, prefix); ok { + *target, err = strconv.ParseInt(strings.TrimSpace(value), 10, 64) + if err != nil { + return "failed", err + } + } + } + } + if loose < 0 || packs < 0 { + return "failed", fmt.Errorf("git count-objects did not report loose objects and packs") + } + // Fetched generations may arrive loose or, with fetch.unpackLimit or a + // large catch-up, as self-contained packs with duplicate delta bases. + // Only a single pack with nothing loose is already consolidated. + if loose == 0 && packs <= 1 { + return "current", nil + } + if loose > portableRepackLooseLimit { + return "skipped-loose-backlog", nil + } + if err := runPortableGit(ctx, root, io.Discard, "repack", "-d", "-q", "--geometric=2", "--no-write-bitmap-index"); err != nil { + return "failed", err + } + return "geometric", nil } func (a *App) runPortableRefresh(ctx context.Context, args []string) error { @@ -329,6 +379,25 @@ func (a *App) refreshPortable(ctx context.Context, options portableRefreshOption } } } + // Only split stores are repacked: their generations delta against each + // other. Raw and gzip subscribers keep the previous no-repack contract. + // This also runs when refs already match, because an implicit reader + // fast-forward can fetch split objects without consolidating them. + if portableSplitArchive(manifest) { + stage("repack") + repack, err := consolidatePortableObjects(ctx, root) + result.Repack = repack + // The generation is already validated and promoted. A consolidation + // failure is reported, not treated as a refused refresh, unless the + // operation's own deadline or growth budget ended it. + if err != nil && ctx.Err() != nil { + return result, err + } + if err != nil { + result.RepackReason = portableRefreshReason(err) + "; the validated generation is unchanged; check that the store's .git/objects directory is writable, then rerun refresh" + fmt.Fprintf(a.Stderr, "gitcrawl: portable refresh: repack failed: %s\n", result.RepackReason) + } + } if err := budget.check(ctx); err != nil { return result, err } @@ -368,25 +437,44 @@ func stagePortableCommit(ctx context.Context, root, staging, relative string, tr if manifest.OutputPath != relative && manifest.OutputPath != filepath.Base(relative) { return "", manifest, fmt.Errorf("manifest outputPath does not match the configured logical database") } - artifactRelative, err := filepath.Rel(staging, artifact) - if err != nil || !pathWithin(staging, artifact) { - return "", manifest, fmt.Errorf("artifact path escapes staging") + artifacts := []string{artifact} + if portableSplitArchive(manifest) { + artifacts = portableArchivePartPaths(artifact, manifest) } - entry, ok := tree[filepath.ToSlash(artifactRelative)] - if !ok { - return "", manifest, fmt.Errorf("manifest artifact is absent from fetched commit") + entries := make([]portableTreeEntry, 0, len(artifacts)) + var fetchedBytes int64 + for index, path := range artifacts { + relativePath, err := filepath.Rel(staging, path) + if err != nil || !pathWithin(staging, path) { + return "", manifest, fmt.Errorf("artifact path escapes staging") + } + entry, ok := tree[filepath.ToSlash(relativePath)] + if !ok { + return "", manifest, fmt.Errorf("manifest artifact is absent from fetched commit") + } + expected := manifest.OutputBytes + switch { + case portableSplitArchive(manifest): + expected = manifest.ArchiveParts[index].Bytes + case compressed: + expected = manifest.ArchiveBytes + } + if entry.size != expected { + return "", manifest, fmt.Errorf("manifest artifact size does not match fetched blob") + } + fetchedBytes += entry.size + entries = append(entries, entry) } - // Allow the extracted blob, inflated DB and both semantic-identity copies, + // Allow the extracted blobs, inflated DB and both semantic-identity copies, // plus another artifact-sized checkout allocation. The monitor also counts // fetched objects and all other positive growth throughout the operation. - if manifest.OutputBytes <= 0 || manifest.OutputBytes > growth/4 || entry.size > (growth-4*manifest.OutputBytes)/2 { + if manifest.OutputBytes <= 0 || manifest.OutputBytes > growth/4 || fetchedBytes > (growth-4*manifest.OutputBytes)/2 { return "", manifest, fmt.Errorf("artifact validation would exceed growth budget") } - if compressed && entry.size != manifest.ArchiveBytes || !compressed && entry.size != manifest.OutputBytes { - return "", manifest, fmt.Errorf("manifest artifact size does not match fetched blob") - } - if err := extractPortableBlob(ctx, root, entry, artifact); err != nil { - return "", manifest, err + for index, entry := range entries { + if err := extractPortableBlob(ctx, root, entry, artifacts[index]); err != nil { + return "", manifest, err + } } stagedDB, err := stagePortableSQLiteSourceTempContext(ctx, logical, filepath.Join(staging, "runtime.db"), 0o600) if err != nil { diff --git a/internal/cli/portable_refresh_test.go b/internal/cli/portable_refresh_test.go index d935500d..7a48c262 100644 --- a/internal/cli/portable_refresh_test.go +++ b/internal/cli/portable_refresh_test.go @@ -242,7 +242,7 @@ func TestPortableRefreshSuccessNoopAndLocalMirror(t *testing.T) { if err != nil || updated.Result != "updated" || updated.BeforeCommit == updated.AfterCommit || updated.SHA256 == "" { t.Fatalf("update: %+v %v", updated, err) } - if updated.Capacity.PeakGrowth <= 0 || updated.Mirror != fixture.mirror { + if updated.Capacity.PeakGrowth <= 0 || updated.Mirror != fixture.mirror || updated.Repack != "" { t.Fatalf("observations: %+v", updated) } if err := sqliteStoreImmutableHealth(context.Background(), fixture.mirror); err != nil { diff --git a/internal/cli/portable_runtime.go b/internal/cli/portable_runtime.go index 557c408d..7e8bd1f6 100644 --- a/internal/cli/portable_runtime.go +++ b/internal/cli/portable_runtime.go @@ -381,13 +381,23 @@ func recentPortableRefresh(value string, now time.Time, maxAge time.Duration) bo } func portableRuntimeNeedsCopy(sourceDBPath, mirrorPath string) (bool, error) { - sourcePath, _, _, err := portableSourceArtifact(sourceDBPath) + sourcePath, manifest, _, err := portableSourceArtifact(sourceDBPath) if err != nil { return false, err } - sourceInfo, err := os.Stat(sourcePath) - if err != nil { - return false, fmt.Errorf("stat portable source db: %w", err) + sourcePaths := []string{sourcePath} + if portableSplitArchive(manifest) { + sourcePaths = portableArchivePartPaths(sourcePath, manifest) + } + var sourceInfo os.FileInfo + for _, path := range sourcePaths { + info, err := os.Stat(path) + if err != nil { + return false, fmt.Errorf("stat portable source db: %w", err) + } + if sourceInfo == nil || info.ModTime().After(sourceInfo.ModTime()) { + sourceInfo = info + } } mirrorInfo, err := os.Stat(mirrorPath) if err != nil { diff --git a/internal/cli/portable_split_test.go b/internal/cli/portable_split_test.go new file mode 100644 index 00000000..723bfd54 --- /dev/null +++ b/internal/cli/portable_split_test.go @@ -0,0 +1,374 @@ +package cli + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + "testing" +) + +// portableTestSplitManifest replaces any raw database, gzip archive or older +// parts beside path with a split archive of partBytes-sized parts. +func portableTestSplitManifest(t *testing.T, path, relative string, partBytes int) { + t.Helper() + data, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + digest, err := fileSHA256(path) + if err != nil { + t.Fatal(err) + } + old, _ := filepath.Glob(path + ".part-*") + for _, stale := range append(old, path+".gz") { + if err := os.Remove(stale); err != nil && !os.IsNotExist(err) { + t.Fatal(err) + } + } + manifest := portableDBManifest{Schema: "gitcrawl-portable-sync-v2", OutputPath: relative, OutputBytes: int64(len(data)), SHA256: fmt.Sprintf("%x", digest), QuickCheck: "ok", + Compression: "split", ArchiveBytes: int64(len(data)), ArchiveSHA256: fmt.Sprintf("%x", digest), MaxArchiveBytes: int64(partBytes)} + for index := 0; index*partBytes < len(data); index++ { + end := min((index+1)*partBytes, len(data)) + name := fmt.Sprintf("%s.part-%03d", filepath.Base(path), index) + if err := os.WriteFile(filepath.Join(filepath.Dir(path), name), data[index*partBytes:end], 0o644); err != nil { + t.Fatal(err) + } + partDigest, err := fileSHA256(filepath.Join(filepath.Dir(path), name)) + if err != nil { + t.Fatal(err) + } + manifest.ArchiveParts = append(manifest.ArchiveParts, portableArchivePart{Path: name, Bytes: int64(end - index*partBytes), SHA256: fmt.Sprintf("%x", partDigest)}) + } + if err := os.Remove(path); err != nil { + t.Fatal(err) + } + encoded, err := json.Marshal(manifest) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(portableDBManifestPath(path), encoded, 0o644); err != nil { + t.Fatal(err) + } +} + +func (fixture portableRefreshFixture) advanceSplit(t *testing.T, number int, body string) { + t.Helper() + path := filepath.Join(fixture.remote, fixture.relative) + if _, err := os.Stat(path); os.IsNotExist(err) { + temp, err := stagePortableSQLiteSourceTemp(path, path, 0o644) + if err != nil { + t.Fatal(err) + } + if err := os.Rename(temp, path); err != nil { + t.Fatal(err) + } + } + seedPortableThread(t, path, number, body) + info, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + portableTestSplitManifest(t, path, fixture.relative, int(info.Size()/4+1)) + portableTestCommit(t, fixture.remote) +} + +func TestPortableRefreshTransitionsGzipToSplitArchive(t *testing.T) { + fixture := newPortableRefreshFixture(t, true) + if result, err := fixture.refresh(t); err != nil || result.MirrorResult != "promoted" { + t.Fatalf("initial gzip: %+v %v", result, err) + } + for generation, body := range []string{"split transition", "split successor"} { + fixture.advanceSplit(t, 10+generation, body) + result, err := fixture.refresh(t) + if err != nil || result.Result != "updated" || result.MirrorResult != "promoted" { + t.Fatalf("split generation %d: %+v %v", generation, result, err) + } + manifest, ok, err := readPortableDBManifest(portableDBManifestPath(filepath.Join(fixture.checkout, fixture.relative))) + if err != nil || !ok || manifest.Compression != "split" || len(manifest.ArchiveParts) < 2 { + t.Fatalf("checkout manifest: %+v ok=%v err=%v", manifest, ok, err) + } + if sum, _ := fileSHA256(fixture.mirror); fmt.Sprintf("%x", sum) != manifest.SHA256 || result.SHA256 != manifest.SHA256 { + t.Fatalf("mirror does not match reassembled split archive") + } + if result.Repack != "geometric" { + t.Fatalf("split generation %d repack = %q", generation, result.Repack) + } + if loose := portableTestGit(t, fixture.checkout, "count-objects", "-v"); !strings.Contains(loose, "count: 0\n") { + t.Fatalf("refresh left loose objects: %s", loose) + } + if _, err := os.Stat(filepath.Join(fixture.checkout, fixture.relative) + ".gz"); !os.IsNotExist(err) { + t.Fatalf("retired gzip archive remains in checkout: %v", err) + } + } + // A reader-style fast-forward leaves fetched split objects loose; the next + // strict refresh is a no-op for refs but still consolidates them. + fixture.advanceSplit(t, 15, "reader fast-forward generation") + portableTestGit(t, fixture.checkout, "fetch", "-q", "origin", "main") + portableTestGit(t, fixture.checkout, "merge", "-q", "--ff-only", "FETCH_HEAD") + portableTestGit(t, fixture.checkout, "update-ref", "refs/remotes/origin/main", "FETCH_HEAD") + if loose := portableTestGit(t, fixture.checkout, "count-objects", "-v"); strings.Contains(loose, "count: 0\n") { + t.Fatalf("fixture fast-forward did not leave loose objects: %s", loose) + } + if result, err := fixture.refresh(t); err != nil || result.Result != "no-op" || result.Repack != "geometric" { + t.Fatalf("no-op consolidation: %+v %v", result, err) + } + if loose := portableTestGit(t, fixture.checkout, "count-objects", "-v"); !strings.Contains(loose, "count: 0\n") { + t.Fatalf("no-op refresh left loose objects: %s", loose) + } + if result, err := fixture.refresh(t); err != nil || result.Result != "no-op" || result.Repack == "failed" || result.Repack == "" { + t.Fatalf("settled split refresh: %+v %v", result, err) + } + // Generations kept as fetched packs (fetch.unpackLimit=1) leave nothing + // loose; repeated packed fetches must still be folded together. + portableTestGit(t, fixture.checkout, "config", "fetch.unpackLimit", "1") + for generation := 0; generation < 3; generation++ { + fixture.advanceSplit(t, 40+generation, fmt.Sprintf("packed fetch generation %d", generation)) + result, err := fixture.refresh(t) + if err != nil || result.Result != "updated" || result.Repack != "geometric" { + t.Fatalf("packed fetch generation %d: %+v %v", generation, result, err) + } + } + objects := portableTestGit(t, fixture.checkout, "count-objects", "-v") + if !strings.Contains(objects, "count: 0\n") || strings.Contains(objects, "packs: 4") || strings.Contains(objects, "packs: 5") { + t.Fatalf("packed fetches were not consolidated: %s", objects) + } + portableTestGit(t, fixture.checkout, "repack", "-a", "-d", "-q") + if result, err := fixture.refresh(t); err != nil || result.Result != "no-op" || result.Repack != "current" { + t.Fatalf("single-pack split refresh: %+v %v", result, err) + } + portableTestGit(t, fixture.checkout, "config", "--unset", "fetch.unpackLimit") + head := portableTestGit(t, fixture.checkout, "rev-parse", "HEAD") + // A tampered part is refused before the checkout or mirror advance. + path := filepath.Join(fixture.remote, fixture.relative) + fixture.advanceSplit(t, 20, "tampered generation") + if err := os.WriteFile(path+".part-001", []byte("tampered"), 0o644); err != nil { + t.Fatal(err) + } + portableTestCommit(t, fixture.remote) + before, _ := fileSHA256(fixture.mirror) + if result, err := fixture.refresh(t); err == nil || result.Result == "updated" { + t.Fatalf("tampered split accepted: %+v", result) + } + if after, _ := fileSHA256(fixture.mirror); after != before { + t.Fatal("refused split refresh replaced the mirror") + } + if got := portableTestGit(t, fixture.checkout, "rev-parse", "HEAD"); got != head { + t.Fatal("refused split refresh advanced the checkout") + } +} + +func TestSplitPortableSourceInflatesAndRejectsInvalidParts(t *testing.T) { + ctx := context.Background() + dir := t.TempDir() + sourceDB := filepath.Join(dir, "checkout", "data", "openclaw__openclaw.sync.db") + mirrorDB := filepath.Join(dir, "runtime", "openclaw__openclaw.sync.db") + seedPortableThread(t, sourceDB, 7, "split portable source") + info, err := os.Stat(sourceDB) + if err != nil { + t.Fatal(err) + } + portableTestSplitManifest(t, sourceDB, "data/openclaw__openclaw.sync.db", int(info.Size()/3+1)) + changed, err := refreshPortableRuntimeDB(ctx, sourceDB, mirrorDB, false, filepath.Join(dir, "config.toml")) + if err != nil || !changed { + t.Fatalf("refresh split portable source: changed=%v err=%v", changed, err) + } + manifest, _, err := readPortableDBManifest(portableDBManifestPath(sourceDB)) + if err != nil { + t.Fatal(err) + } + if sum, _ := fileSHA256(mirrorDB); fmt.Sprintf("%x", sum) != manifest.SHA256 { + t.Fatal("inflated split mirror does not match manifest") + } + if err := validatePortableSQLiteSourceFile(ctx, sourceDB, sourceDB); err != nil { + t.Fatalf("validate split source: %v", err) + } + for _, tc := range []struct { + name string + mutate func(*portableDBManifest) + message string + }{ + {"escaping part", func(m *portableDBManifest) { m.ArchiveParts[0].Path = "../escape" }, "unique basename"}, + {"duplicate part", func(m *portableDBManifest) { m.ArchiveParts[1].Path = m.ArchiveParts[0].Path }, "unique basename"}, + {"missing parts", func(m *portableDBManifest) { m.ArchiveParts = nil }, "parts"}, + {"total mismatch", func(m *portableDBManifest) { m.ArchiveParts = m.ArchiveParts[:1] }, "total"}, + {"archive digest", func(m *portableDBManifest) { m.ArchiveSHA256 = strings.Repeat("0", 64) }, "equal the database"}, + {"part budget", func(m *portableDBManifest) { m.MaxArchiveBytes = 1 }, "budget"}, + {"single archive path", func(m *portableDBManifest) { m.ArchivePath = "x.gz" }, "archivePath"}, + {"parts without split", func(m *portableDBManifest) { m.Compression = "" }, "require split"}, + {"parts with gzip", func(m *portableDBManifest) { m.Compression = "gzip" }, "require split"}, + } { + t.Run(tc.name, func(t *testing.T) { + changed := manifest + changed.ArchiveParts = append([]portableArchivePart(nil), manifest.ArchiveParts...) + tc.mutate(&changed) + encoded, _ := json.Marshal(changed) + if err := os.WriteFile(portableDBManifestPath(sourceDB), encoded, 0o644); err != nil { + t.Fatal(err) + } + _, _, _, err := portableSourceArtifact(sourceDB) + if err == nil || !strings.Contains(err.Error(), tc.message) { + t.Fatalf("error = %v, want %q", err, tc.message) + } + }) + } + encoded, _ := json.Marshal(manifest) + if err := os.WriteFile(portableDBManifestPath(sourceDB), encoded, 0o644); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(filepath.Dir(sourceDB), manifest.ArchiveParts[1].Path), []byte("x"), 0o644); err != nil { + t.Fatal(err) + } + if _, err := refreshPortableRuntimeDB(ctx, sourceDB, mirrorDB, true, filepath.Join(dir, "config.toml")); err == nil || !strings.Contains(err.Error(), "portable manifest mismatch") { + t.Fatalf("tampered part error = %v", err) + } + if backup, err := preserveMalformedPortableDB(filepath.Join(dir, "checkout"), sourceDB); err == nil { + for _, part := range manifest.ArchiveParts { + if _, err := os.Stat(filepath.Join(backup, part.Path+".malformed")); err != nil { + t.Fatalf("malformed evidence missing part %s: %v", part.Path, err) + } + } + } else { + t.Fatalf("preserve malformed split source: %v", err) + } +} + +func TestConsolidatePortableObjectsSkipsLargeLooseBacklog(t *testing.T) { + fixture := newPortableRefreshFixture(t, false) + objects := filepath.Join(fixture.checkout, ".git", "objects") + for index := 0; index <= portableRepackLooseLimit; index++ { + path := filepath.Join(fixture.checkout, fmt.Sprintf("loose-%d", index)) + if err := os.WriteFile(path, []byte(fmt.Sprintf("loose object %d", index)), 0o600); err != nil { + t.Fatal(err) + } + portableTestGit(t, fixture.checkout, "hash-object", "-w", path) + if err := os.Remove(path); err != nil { + t.Fatal(err) + } + } + before, _ := filepath.Glob(filepath.Join(objects, "pack", "*.pack")) + ctx, err := portableGitContext(context.Background(), "") + if err != nil { + t.Fatal(err) + } + result, err := consolidatePortableObjects(ctx, fixture.checkout) + if err != nil || result != "skipped-loose-backlog" { + t.Fatalf("backlog result=%q err=%v", result, err) + } + after, _ := filepath.Glob(filepath.Join(objects, "pack", "*.pack")) + if len(after) != len(before) { + t.Fatal("skipped consolidation wrote a pack") + } +} + +func TestPortableStatusReadsSplitArchiveWithoutRuntime(t *testing.T) { + fixture := newPortableRefreshFixture(t, true) + fixture.advanceSplit(t, 30, "split status generation") + if result, err := fixture.refresh(t); err != nil || result.Result != "updated" { + t.Fatalf("split refresh: %+v %v", result, err) + } + if err := os.Remove(fixture.mirror); err != nil { + t.Fatal(err) + } + logical := filepath.Join(fixture.checkout, fixture.relative) + output := fixture.command(t, "status", "--json") + var status struct { + DatabasePath string `json:"database_path"` + Databases []struct { + Kind string `json:"kind"` + } `json:"databases"` + Warnings []string `json:"warnings"` + } + if err := json.Unmarshal(output, &status); err != nil { + t.Fatal(err) + } + if !sameExistingPath(status.DatabasePath, logical+".part-000") || len(status.Databases) == 0 || status.Databases[0].Kind != "sqlite-split" { + t.Fatalf("split status inventory: %s", output) + } + if !strings.Contains(strings.Join(status.Warnings, " "), "split artifact") { + t.Fatalf("split status warning missing: %s", output) + } + if _, err := os.Stat(logical); !os.IsNotExist(err) { + t.Fatalf("status materialized logical database: %v", err) + } +} + +func TestSplitPortableArchiveValidationRejectsMissingAndResizedParts(t *testing.T) { + ctx := context.Background() + dir := t.TempDir() + sourceDB := filepath.Join(dir, "data", "archive.db") + seedPortableThread(t, sourceDB, 4, "split validation") + info, err := os.Stat(sourceDB) + if err != nil { + t.Fatal(err) + } + portableTestSplitManifest(t, sourceDB, "data/archive.db", int(info.Size()/2+1)) + first, manifest, compressed, err := portableSourceArtifact(sourceDB) + if err != nil || !compressed { + t.Fatalf("split source: compressed=%v err=%v", compressed, err) + } + if err := validatePortableArchive(first, manifest); err != nil { + t.Fatalf("valid split archive: %v", err) + } + second := portableArchivePartPaths(first, manifest)[1] + data, err := os.ReadFile(second) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(second, append(data, 0), 0o644); err != nil { + t.Fatal(err) + } + if err := validatePortableArchiveContext(ctx, first, manifest); err == nil || !strings.Contains(err.Error(), "size") { + t.Fatalf("resized part error = %v", err) + } + if err := os.Remove(second); err != nil { + t.Fatal(err) + } + if err := validatePortableArchiveContext(ctx, first, manifest); err == nil { + t.Fatal("missing part accepted") + } + if _, err := stagePortableSQLiteSourceTempContext(ctx, sourceDB, filepath.Join(dir, "runtime", "archive.db"), 0o600); err == nil { + t.Fatal("staged a split archive with a missing part") + } + tampered := manifest + tampered.ArchiveParts = append([]portableArchivePart(nil), manifest.ArchiveParts...) + tampered.ArchiveParts[0].SHA256 = "short" + if err := validatePortableSplitMetadata(tampered); err == nil || !strings.Contains(err.Error(), "sha256") { + t.Fatalf("short part digest error = %v", err) + } + tampered.ArchiveParts[0].SHA256 = manifest.ArchiveParts[0].SHA256 + tampered.ArchiveParts[0].Bytes = manifest.OutputBytes + 1 + if err := validatePortableSplitMetadata(tampered); err == nil { + t.Fatal("oversized part accepted") + } +} + +func TestPortableExportRejectsUnknownPackaging(t *testing.T) { + app := New() + app.Stdout, app.Stderr = &bytes.Buffer{}, &bytes.Buffer{} + err := app.Run(context.Background(), []string{"--config", filepath.Join(t.TempDir(), "missing.toml"), + "portable", "export", "--profile", "current-state-v1", "--output-dir", filepath.Join(t.TempDir(), "out"), "--compression", "zstd"}) + if err == nil || !strings.Contains(err.Error(), `"gzip" or "split"`) { + t.Fatalf("unknown packaging error = %v", err) + } +} + +func TestPortableRefreshReportsRepackFailureWithoutRefusing(t *testing.T) { + fixture := newPortableRefreshFixture(t, true) + fixture.advanceSplit(t, 50, "repack failure generation") + pack := filepath.Join(fixture.checkout, ".git", "objects", "pack") + if err := os.Chmod(pack, 0o555); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = os.Chmod(pack, 0o755) }) + result, err := fixture.refresh(t) + if err != nil || result.Result != "updated" || result.Repack != "failed" || !strings.Contains(result.RepackReason, "writable") { + t.Fatalf("repack failure: %+v %v", result, err) + } + if sum, _ := fileSHA256(fixture.mirror); fmt.Sprintf("%x", sum) != result.SHA256 { + t.Fatal("repack failure changed the validated generation") + } +} diff --git a/internal/cli/portable_staging.go b/internal/cli/portable_staging.go index 202f6923..5ca05aea 100644 --- a/internal/cli/portable_staging.go +++ b/internal/cli/portable_staging.go @@ -113,16 +113,33 @@ func stagePortableSQLiteSourceTempContext(ctx context.Context, sourceDBPath, tar if err := os.MkdirAll(filepath.Dir(targetPath), 0o755); err != nil { return "", fmt.Errorf("create portable runtime dir: %w", err) } - source, err := os.Open(sourcePath) - if err != nil { - return "", fmt.Errorf("open portable source archive: %w", err) - } - defer source.Close() - reader, err := gzip.NewReader(source) - if err != nil { - return "", fmt.Errorf("open portable gzip archive: %w", err) + var reader io.Reader + if portableSplitArchive(manifest) { + // Reassemble the validated parts in order; the whole digest is checked + // again by the caller's manifest validation of the staged database. + var readers []io.Reader + for _, path := range portableArchivePartPaths(sourcePath, manifest) { + part, err := os.Open(path) + if err != nil { + return "", fmt.Errorf("open portable archive part: %w", err) + } + defer part.Close() + readers = append(readers, part) + } + reader = io.MultiReader(readers...) + } else { + source, err := os.Open(sourcePath) + if err != nil { + return "", fmt.Errorf("open portable source archive: %w", err) + } + defer source.Close() + gzipReader, err := gzip.NewReader(source) + if err != nil { + return "", fmt.Errorf("open portable gzip archive: %w", err) + } + defer gzipReader.Close() + reader = gzipReader } - defer reader.Close() temp, err := os.CreateTemp(filepath.Dir(targetPath), "."+filepath.Base(targetPath)+".tmp-*") if err != nil { return "", fmt.Errorf("create portable runtime temp db: %w", err) diff --git a/internal/cli/status.go b/internal/cli/status.go index 47116220..bd06aa4b 100644 --- a/internal/cli/status.go +++ b/internal/cli/status.go @@ -21,6 +21,7 @@ func (a *App) localArchiveStatus(ctx context.Context, cfg config.Config) (contro var warnings []string stale := false immutable := false + splitArtifact := false runtimePath := "" _, portable, err := portableStoreRoot(ctx, cfg.DBPath) if err != nil { @@ -46,7 +47,7 @@ func (a *App) localArchiveStatus(ctx context.Context, cfg config.Config) (contro } else if !errors.Is(err, os.ErrNotExist) { return control.Status{}, err } else { - artifact, _, compressed, err := portableSourceArtifact(cfg.DBPath) + artifact, artifactManifest, compressed, err := portableSourceArtifact(cfg.DBPath) if err != nil { return control.Status{}, err } @@ -64,7 +65,12 @@ func (a *App) localArchiveStatus(ctx context.Context, cfg config.Config) (contro return control.Status{}, err } reportedPath, immutable = artifact, true - warnings = append(warnings, "No runtime mirror exists; counts describe the validated gzip artifact and database bytes are its compressed on-disk size.") + if portableSplitArchive(artifactManifest) { + splitArtifact = true + warnings = append(warnings, "No runtime mirror exists; counts describe the validated split artifact and database bytes are its first part's on-disk size.") + } else { + warnings = append(warnings, "No runtime mirror exists; counts describe the validated gzip artifact and database bytes are its compressed on-disk size.") + } } } } @@ -103,6 +109,9 @@ func (a *App) localArchiveStatus(ctx context.Context, cfg config.Config) (contro out.Warnings = append(out.Warnings, warnings...) if immutable { out.Databases[0].Kind = "sqlite-gzip" + if splitArtifact { + out.Databases[0].Kind = "sqlite-split" + } } return out, nil } diff --git a/internal/portable/artifact_io.go b/internal/portable/artifact_io.go index 9eaa3c51..a14db3ee 100644 --- a/internal/portable/artifact_io.go +++ b/internal/portable/artifact_io.go @@ -8,6 +8,7 @@ import ( "fmt" "io" "os" + "path/filepath" ) func fileSHA256(filePath string) (string, error) { @@ -158,3 +159,111 @@ func removeSQLiteSidecars(dbPath string) error { } return nil } + +// splitPartName names part index i of a split archive beside its database. +func splitPartName(dbName string, index int) string { + return fmt.Sprintf("%s.part-%03d", dbName, index) +} + +// writeSplitArchive copies the database into ordered parts of at most +// partBytes. Each part is written exclusively and synced; a failure removes +// every part it created. +func writeSplitArchive(dbPath string, partBytes int64) (parts []ArchivePart, retErr error) { + if partBytes <= 0 { + return nil, fmt.Errorf("split part size must be positive") + } + source, err := os.Open(dbPath) + if err != nil { + return nil, err + } + defer source.Close() + info, err := source.Stat() + if err != nil { + return nil, err + } + if info.Size() <= 0 { + return nil, fmt.Errorf("split archive requires a non-empty database") + } + count := (info.Size() + partBytes - 1) / partBytes + if count > maxArchiveParts { + return nil, fmt.Errorf("split archive needs %d parts; maximum is %d", count, maxArchiveParts) + } + dir := filepath.Dir(dbPath) + defer func() { + if retErr != nil { + for _, part := range parts { + _ = os.Remove(filepath.Join(dir, part.Path)) + } + parts = nil + } + }() + for index := 0; int64(index) < count; index++ { + name := splitPartName(filepath.Base(dbPath), index) + file, err := os.OpenFile(filepath.Join(dir, name), os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o644) + if err != nil { + return parts, err + } + parts = append(parts, ArchivePart{Path: name}) + hash := sha256.New() + written, copyErr := io.Copy(io.MultiWriter(file, hash), io.LimitReader(source, partBytes)) + syncErr := file.Sync() + closeErr := file.Close() + if err := errors.Join(copyErr, syncErr, closeErr); err != nil { + return parts, err + } + if written <= 0 { + return parts, fmt.Errorf("split archive part %d is empty", index) + } + parts[index].Bytes = written + parts[index].SHA256 = hex.EncodeToString(hash.Sum(nil)) + } + return parts, nil +} + +// validateSplitArchive verifies part names, sizes and digests, and that the +// ordered concatenation is exactly the manifest database. +func validateSplitArchive(dir string, parts []ArchivePart, archiveBytes int64, archiveSHA string, outputBytes int64, outputSHA string, maxPartBytes *int64) error { + if len(parts) == 0 || len(parts) > maxArchiveParts { + return fmt.Errorf("portable manifest split archive must have 1..%d parts", maxArchiveParts) + } + if archiveBytes != outputBytes || archiveSHA != outputSHA { + return fmt.Errorf("portable manifest split archive metadata must equal the database size and sha256") + } + whole := sha256.New() + var total int64 + for index, part := range parts { + if part.Path == "" || filepath.Base(part.Path) != part.Path || part.Path == "." || part.Path == ".." { + return fmt.Errorf("portable manifest archive part path must be a basename") + } + if part.Bytes <= 0 || part.SHA256 == "" { + return fmt.Errorf("portable manifest archive part %d metadata is incomplete", index) + } + if maxPartBytes != nil && part.Bytes > *maxPartBytes { + return fmt.Errorf("portable manifest archive part %d exceeds maxArchiveBytes", index) + } + file, err := os.Open(filepath.Join(dir, part.Path)) + if err != nil { + return fmt.Errorf("open portable archive part: %w", err) + } + hash := sha256.New() + written, copyErr := io.Copy(io.MultiWriter(hash, whole), io.LimitReader(file, part.Bytes+1)) + closeErr := file.Close() + if err := errors.Join(copyErr, closeErr); err != nil { + return fmt.Errorf("read portable archive part: %w", err) + } + if written != part.Bytes { + return fmt.Errorf("portable archive part %d has %d bytes, expected %d", index, written, part.Bytes) + } + if hex.EncodeToString(hash.Sum(nil)) != part.SHA256 { + return fmt.Errorf("portable archive part %d does not match its sha256", index) + } + total += written + } + if total != outputBytes { + return fmt.Errorf("portable archive parts total %d bytes, expected %d", total, outputBytes) + } + if hex.EncodeToString(whole.Sum(nil)) != outputSHA { + return fmt.Errorf("portable archive parts do not match database sha256") + } + return nil +} diff --git a/internal/portable/export.go b/internal/portable/export.go index e461d0aa..db86f255 100644 --- a/internal/portable/export.go +++ b/internal/portable/export.go @@ -16,13 +16,21 @@ import ( ) const ( - CurrentStateV1 = "current-state-v1" - CompressionGzip = "gzip" - portableSchema = "gitcrawl-portable-sync-v2" - currentProfileVersion = 1 - defaultBodyChars = 256 - onlineBackupPageChunk = int32(1024) - onlineBackupBusyRetry = 50 * time.Millisecond + CurrentStateV1 = "current-state-v1" + CompressionGzip = "gzip" + // CompressionSplit stores the uncompressed SQLite bytes as ordered parts. + // Consecutive generations then share most content, so Git's delta + // compression stores each new generation as a small change instead of a new + // fully compressed blob that can never delta against its predecessor. + CompressionSplit = "split" + // SplitPartBytes keeps each committed part far below hosting file limits. + SplitPartBytes int64 = 32 << 20 + maxArchiveParts = 64 + portableSchema = "gitcrawl-portable-sync-v2" + currentProfileVersion = 1 + defaultBodyChars = 256 + onlineBackupPageChunk = int32(1024) + onlineBackupBusyRetry = 50 * time.Millisecond ) type ExportOptions struct { @@ -60,38 +68,39 @@ const ( type ProgressFunc func(Stage) type ExportResult struct { - Profile string `json:"profile"` - 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"` - PublicPath string `json:"public_path"` - Repository *Repository `json:"repository,omitempty"` - BodyChars int `json:"body_chars"` - BytesBefore int64 `json:"bytes_before"` - BytesAfter int64 `json:"bytes_after"` - MaxBytes *int64 `json:"max_bytes,omitempty"` - ByteBudgetOK bool `json:"byte_budget_ok"` - Compression string `json:"compression,omitempty"` - ArchivePath string `json:"archive_path,omitempty"` - ArchiveBytes int64 `json:"archive_bytes,omitempty"` - ArchiveSHA256 string `json:"archive_sha256,omitempty"` - MaxArchiveBytes *int64 `json:"max_archive_bytes,omitempty"` - ArchiveByteBudgetOK bool `json:"archive_byte_budget_ok"` - ArtifactID string `json:"artifact_id"` - ArtifactIDProfile string `json:"artifact_id_profile"` - SHA256 string `json:"sha256"` - QuickCheck string `json:"quick_check"` - IntegrityCheck string `json:"integrity_check"` - ForeignKeyViolations int `json:"foreign_key_violations"` - Vacuumed bool `json:"vacuumed"` - DroppedTables []string `json:"dropped_tables"` - DroppedIndexes []string `json:"dropped_indexes"` - ArtifactCommitted bool `json:"artifact_committed"` - ColumnProfile string `json:"column_profile,omitempty"` + Profile string `json:"profile"` + 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"` + PublicPath string `json:"public_path"` + Repository *Repository `json:"repository,omitempty"` + BodyChars int `json:"body_chars"` + BytesBefore int64 `json:"bytes_before"` + BytesAfter int64 `json:"bytes_after"` + MaxBytes *int64 `json:"max_bytes,omitempty"` + ByteBudgetOK bool `json:"byte_budget_ok"` + Compression string `json:"compression,omitempty"` + ArchivePath string `json:"archive_path,omitempty"` + ArchiveParts []ArchivePart `json:"archive_parts,omitempty"` + ArchiveBytes int64 `json:"archive_bytes,omitempty"` + ArchiveSHA256 string `json:"archive_sha256,omitempty"` + MaxArchiveBytes *int64 `json:"max_archive_bytes,omitempty"` + ArchiveByteBudgetOK bool `json:"archive_byte_budget_ok"` + ArtifactID string `json:"artifact_id"` + ArtifactIDProfile string `json:"artifact_id_profile"` + SHA256 string `json:"sha256"` + QuickCheck string `json:"quick_check"` + IntegrityCheck string `json:"integrity_check"` + ForeignKeyViolations int `json:"foreign_key_violations"` + Vacuumed bool `json:"vacuumed"` + DroppedTables []string `json:"dropped_tables"` + DroppedIndexes []string `json:"dropped_indexes"` + ArtifactCommitted bool `json:"artifact_committed"` + ColumnProfile string `json:"column_profile,omitempty"` } type exporter struct { @@ -121,8 +130,8 @@ func (e exporter) export(ctx context.Context, options ExportOptions) (result Exp if options.MaxBytes != nil && *options.MaxBytes <= 0 { return result, fmt.Errorf("max-bytes must be a positive integer") } - if options.Compression != "" && options.Compression != CompressionGzip { - return result, fmt.Errorf("unsupported compression %q; supported compression: %s", options.Compression, CompressionGzip) + if options.Compression != "" && options.Compression != CompressionGzip && options.Compression != CompressionSplit { + return result, fmt.Errorf("unsupported compression %q; supported compression: %s, %s", options.Compression, CompressionGzip, CompressionSplit) } if options.MaxArchiveBytes != nil { if *options.MaxArchiveBytes <= 0 { @@ -191,6 +200,7 @@ func (e exporter) export(ctx context.Context, options ExportOptions) (result Exp if archivePath != "" { result.ArchivePath = filepath.Join(outputDir, filepath.Base(archivePath)) } + var archiveParts []ArchivePart if err := reportProgress(ctx, options.Progress, StageSnapshot); err != nil { return result, err } @@ -422,6 +432,26 @@ func (e exporter) export(ctx context.Context, options ExportOptions) (result Exp return result, fmt.Errorf("hash portable archive: %w", err) } } + if options.Compression == CompressionSplit { + if err := reportProgress(ctx, options.Progress, StageCompression); err != nil { + return result, err + } + partBytes := SplitPartBytes + if options.MaxArchiveBytes != nil && *options.MaxArchiveBytes < partBytes { + partBytes = *options.MaxArchiveBytes + } + archiveParts, err = writeSplitArchive(dbPath, partBytes) + if err != nil { + return result, fmt.Errorf("split portable database: %w", err) + } + // The parts are the database bytes; their concatenation has the same + // size and digest, which readers verify after reassembly. + result.ArchiveBytes = info.Size() + result.ArchiveSHA256 = sha + for _, part := range archiveParts { + result.ArchiveParts = append(result.ArchiveParts, ArchivePart{Path: filepath.Join(outputDir, part.Path), Bytes: part.Bytes, SHA256: part.SHA256}) + } + } if err := reportProgress(ctx, options.Progress, StageManifest); err != nil { return result, err } @@ -457,6 +487,9 @@ func (e exporter) export(ctx context.Context, options ExportOptions) (result Exp if archivePath != "" { manifest.ArchivePath = filepath.Base(archivePath) } + if options.Compression == CompressionSplit { + manifest.ArchiveParts = archiveParts + } if e.beforeManifest != nil { if err := e.beforeManifest(); err != nil { return result, err @@ -477,6 +510,11 @@ func (e exporter) export(ctx context.Context, options ExportOptions) (result Exp return result, fmt.Errorf("sync portable archive: %w", err) } } + for _, part := range archiveParts { + if err := syncRegularFile(filepath.Join(stageDir, part.Path)); err != nil { + return result, fmt.Errorf("sync portable archive part: %w", err) + } + } if err := validateManifestPair(ctx, dbPath, manifestPath, manifest); err != nil { return result, err } @@ -488,7 +526,7 @@ func (e exporter) export(ctx context.Context, options ExportOptions) (result Exp } else if finalSHA != sha { return result, fmt.Errorf("portable database changed during manifest validation") } - if archivePath != "" { + if archivePath != "" || len(archiveParts) > 0 { if err := os.Remove(dbPath); err != nil { return result, fmt.Errorf("remove unpackaged portable database: %w", err) } diff --git a/internal/portable/export_split_test.go b/internal/portable/export_split_test.go new file mode 100644 index 00000000..9f24baed --- /dev/null +++ b/internal/portable/export_split_test.go @@ -0,0 +1,156 @@ +package portable + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "os" + "path/filepath" + "strings" + "testing" +) + +func TestExportSplitCommitsVerifiedOrderedParts(t *testing.T) { + ctx := context.Background() + dir := t.TempDir() + sourcePath := filepath.Join(dir, "source.db") + st := seedExportSource(t, ctx, sourcePath) + defer st.Close() + + baseline, err := Export(ctx, testExportOptions(sourcePath, filepath.Join(dir, "baseline"))) + if err != nil { + t.Fatalf("baseline export: %v", err) + } + partBudget := baseline.BytesAfter/3 + 1 + options := testExportOptions(sourcePath, filepath.Join(dir, "split")) + options.Compression = CompressionSplit + options.MaxArchiveBytes = &partBudget + split, err := Export(ctx, options) + if err != nil { + t.Fatalf("split export: %v", err) + } + if split.Compression != CompressionSplit || split.ArchivePath != "" || len(split.ArchiveParts) != 3 || + split.ArchiveBytes != split.BytesAfter || split.ArchiveSHA256 != split.SHA256 { + t.Fatalf("split result = %+v", split) + } + if _, err := os.Stat(split.DatabasePath); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("split export retained raw database: %v", err) + } + manifest := readManifest(t, split.ManifestPath) + if manifest.Compression != CompressionSplit || manifest.ArchivePath != "" || len(manifest.ArchiveParts) != 3 || + manifest.ArchiveBytes != split.BytesAfter || manifest.ArchiveSHA256 != split.SHA256 || + manifest.MaxArchiveBytes == nil || *manifest.MaxArchiveBytes != partBudget { + t.Fatalf("split manifest = %+v", manifest) + } + var whole bytes.Buffer + for index, part := range manifest.ArchiveParts { + if part.Path != splitPartName(filepath.Base(split.DatabasePath), index) || part.Bytes > partBudget || + split.ArchiveParts[index].Path != filepath.Join(split.OutputDir, part.Path) { + t.Fatalf("part %d = %+v result=%+v", index, part, split.ArchiveParts[index]) + } + data, err := os.ReadFile(filepath.Join(split.OutputDir, part.Path)) + if err != nil { + t.Fatal(err) + } + sum := sha256.Sum256(data) + if hex.EncodeToString(sum[:]) != part.SHA256 || int64(len(data)) != part.Bytes { + t.Fatalf("part %d digest/size mismatch", index) + } + whole.Write(data) + } + sum := sha256.Sum256(whole.Bytes()) + if hex.EncodeToString(sum[:]) != split.SHA256 || split.ArtifactID != baseline.ArtifactID { + t.Fatalf("reassembled split database does not match its manifest") + } + if err := validateSplitArchive(split.OutputDir, manifest.ArchiveParts, manifest.ArchiveBytes, manifest.ArchiveSHA256, manifest.OutputBytes, manifest.SHA256, manifest.MaxArchiveBytes); err != nil { + t.Fatalf("validate committed split archive: %v", err) + } + // Tampering, reordering and budget violations are each rejected. + reordered := append([]ArchivePart(nil), manifest.ArchiveParts...) + reordered[0], reordered[1] = reordered[1], reordered[0] + small := int64(1) + for name, check := range map[string]func() error{ + "reordered": func() error { + return validateSplitArchive(split.OutputDir, reordered, manifest.ArchiveBytes, manifest.ArchiveSHA256, manifest.OutputBytes, manifest.SHA256, nil) + }, + "budget": func() error { + return validateSplitArchive(split.OutputDir, manifest.ArchiveParts, manifest.ArchiveBytes, manifest.ArchiveSHA256, manifest.OutputBytes, manifest.SHA256, &small) + }, + "metadata": func() error { + return validateSplitArchive(split.OutputDir, manifest.ArchiveParts, manifest.ArchiveBytes-1, manifest.ArchiveSHA256, manifest.OutputBytes, manifest.SHA256, nil) + }, + } { + if err := check(); err == nil { + t.Fatalf("%s split archive accepted", name) + } + } + lastPart := filepath.Join(split.OutputDir, manifest.ArchiveParts[2].Path) + data, _ := os.ReadFile(lastPart) + data[len(data)-1] ^= 0xff + if err := os.WriteFile(lastPart, data, 0o644); err != nil { + t.Fatal(err) + } + if err := validateSplitArchive(split.OutputDir, manifest.ArchiveParts, manifest.ArchiveBytes, manifest.ArchiveSHA256, manifest.OutputBytes, manifest.SHA256, nil); err == nil || !strings.Contains(err.Error(), "sha256") { + t.Fatalf("tampered split archive error = %v", err) + } + + tiny := int64(1024) + failingDir := filepath.Join(dir, "too-many-parts") + failing := testExportOptions(sourcePath, failingDir) + failing.Compression = CompressionSplit + failing.MaxArchiveBytes = &tiny + if _, err := Export(ctx, failing); err == nil || !strings.Contains(err.Error(), "parts") { + t.Fatalf("too many parts error = %v", err) + } + if _, err := os.Stat(failingDir); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("failed split export left target: %v", err) + } + assertNoExportTemps(t, dir) +} + +func TestWriteSplitArchiveRefusesInvalidInputsAndCleansUp(t *testing.T) { + dir := t.TempDir() + empty := filepath.Join(dir, "empty.db") + if err := os.WriteFile(empty, nil, 0o600); err != nil { + t.Fatal(err) + } + if _, err := writeSplitArchive(empty, 16); err == nil { + t.Fatal("empty database split") + } + data := filepath.Join(dir, "data.db") + if err := os.WriteFile(data, bytes.Repeat([]byte("x"), 100), 0o600); err != nil { + t.Fatal(err) + } + if _, err := writeSplitArchive(data, 0); err == nil { + t.Fatal("zero part size accepted") + } + if _, err := writeSplitArchive(filepath.Join(dir, "missing.db"), 16); err == nil { + t.Fatal("missing database split") + } + // A pre-existing part blocks exclusive creation; earlier parts are removed. + if err := os.WriteFile(filepath.Join(dir, splitPartName("data.db", 2)), []byte("existing"), 0o600); err != nil { + t.Fatal(err) + } + if _, err := writeSplitArchive(data, 40); err == nil { + t.Fatal("split overwrote an existing part") + } + for _, index := range []int{0, 1} { + if _, err := os.Stat(filepath.Join(dir, splitPartName("data.db", index))); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("failed split left part %d: %v", index, err) + } + } + if _, err := writeSplitArchive(data, 1); err == nil || !strings.Contains(err.Error(), "maximum") { + t.Fatalf("too many parts error = %v", err) + } + if err := validateSplitArchive(dir, []ArchivePart{{Path: "../x", Bytes: 1, SHA256: "a"}}, 1, "s", 1, "s", nil); err == nil { + t.Fatal("escaping part path accepted") + } + if err := validateSplitArchive(dir, []ArchivePart{{Path: "data.db", Bytes: 0}}, 1, "s", 1, "s", nil); err == nil { + t.Fatal("incomplete part metadata accepted") + } + if err := validateSplitArchive(dir, nil, 1, "s", 1, "s", nil); err == nil { + t.Fatal("empty part list accepted") + } +} diff --git a/internal/portable/manifest.go b/internal/portable/manifest.go index 36e25997..b3a60660 100644 --- a/internal/portable/manifest.go +++ b/internal/portable/manifest.go @@ -20,40 +20,49 @@ type Repository struct { FullName string `json:"fullName"` } +// ArchivePart is one ordered piece of a split archive. Path is a basename +// beside the manifest; the concatenation of all parts is the database. +type ArchivePart struct { + Path string `json:"path"` + Bytes int64 `json:"bytes"` + SHA256 string `json:"sha256"` +} + type Table struct { Name string `json:"name"` Rows int64 `json:"rows"` } type Manifest struct { - Schema string `json:"schema"` - PortableSchema string `json:"portableSchema"` - Profile string `json:"profile"` - ProfileVersion int `json:"profileVersion"` - ExportedAt string `json:"exportedAt"` - OutputPath string `json:"outputPath"` - OutputBytes int64 `json:"outputBytes"` - SHA256 string `json:"sha256"` - ArtifactID string `json:"artifactId"` - ArtifactIDProfile string `json:"artifactIdProfile"` - Repository *Repository `json:"repository,omitempty"` - BodyChars int `json:"bodyChars"` - MaxBytes *int64 `json:"maxBytes,omitempty"` - Compression string `json:"compression,omitempty"` - ArchivePath string `json:"archivePath,omitempty"` - ArchiveBytes int64 `json:"archiveBytes,omitempty"` - ArchiveSHA256 string `json:"archiveSha256,omitempty"` - MaxArchiveBytes *int64 `json:"maxArchiveBytes,omitempty"` - Tables []Table `json:"tables"` - Excluded []string `json:"excluded"` - ValidationOK bool `json:"validationOk"` - QuickCheck string `json:"quickCheck"` - IntegrityCheck string `json:"integrityCheck"` - ForeignKeyViolations int `json:"foreignKeyViolations"` - DroppedTables []string `json:"droppedTables"` - DroppedIndexes []string `json:"droppedIndexes"` - IndexProfile string `json:"indexProfile"` - ColumnProfile string `json:"columnProfile,omitempty"` + Schema string `json:"schema"` + PortableSchema string `json:"portableSchema"` + Profile string `json:"profile"` + ProfileVersion int `json:"profileVersion"` + ExportedAt string `json:"exportedAt"` + OutputPath string `json:"outputPath"` + OutputBytes int64 `json:"outputBytes"` + SHA256 string `json:"sha256"` + ArtifactID string `json:"artifactId"` + ArtifactIDProfile string `json:"artifactIdProfile"` + Repository *Repository `json:"repository,omitempty"` + BodyChars int `json:"bodyChars"` + MaxBytes *int64 `json:"maxBytes,omitempty"` + Compression string `json:"compression,omitempty"` + ArchivePath string `json:"archivePath,omitempty"` + ArchiveParts []ArchivePart `json:"archiveParts,omitempty"` + ArchiveBytes int64 `json:"archiveBytes,omitempty"` + ArchiveSHA256 string `json:"archiveSha256,omitempty"` + MaxArchiveBytes *int64 `json:"maxArchiveBytes,omitempty"` + Tables []Table `json:"tables"` + Excluded []string `json:"excluded"` + ValidationOK bool `json:"validationOk"` + QuickCheck string `json:"quickCheck"` + IntegrityCheck string `json:"integrityCheck"` + ForeignKeyViolations int `json:"foreignKeyViolations"` + DroppedTables []string `json:"droppedTables"` + DroppedIndexes []string `json:"droppedIndexes"` + IndexProfile string `json:"indexProfile"` + ColumnProfile string `json:"columnProfile,omitempty"` } func ValidateDatabaseName(name string) error { @@ -115,7 +124,16 @@ func validateManifestPair(ctx context.Context, dbPath, manifestPath string, expe if sha != actual.SHA256 { return fmt.Errorf("portable manifest sha256 does not match database") } - if actual.Compression != "" { + if actual.Compression == CompressionSplit { + if actual.ArchivePath != "" { + return fmt.Errorf("portable manifest split archive must not name a single archivePath") + } + if err := validateSplitArchive(filepath.Dir(manifestPath), actual.ArchiveParts, actual.ArchiveBytes, actual.ArchiveSHA256, actual.OutputBytes, actual.SHA256, actual.MaxArchiveBytes); err != nil { + return err + } + } else if len(actual.ArchiveParts) > 0 { + return fmt.Errorf("portable manifest archiveParts require split compression") + } else if actual.Compression != "" { if actual.Compression != CompressionGzip { return fmt.Errorf("portable manifest compression %q is unsupported", actual.Compression) }