From ebd449d8474fa071901bf9f4a4628fbb02832d77 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sat, 18 Jul 2026 00:07:53 +0000 Subject: [PATCH] Add chunk plan cache configuration and storage --- docs/roadmap/implementation.md | 2 +- internal/core/config/chunk_cache_test.go | 126 ++++++++ internal/core/config/config.go | 7 + internal/core/config/env.go | 11 + internal/core/config/file_config.go | 27 ++ internal/core/config/validation.go | 6 + internal/core/workspace/chunk_plan.go | 22 ++ internal/core/workspace/chunk_plan_test.go | 57 ++++ internal/framework/chunkplan/store.go | 207 +++++++++++++ internal/framework/chunkplan/store_test.go | 283 ++++++++++++++++++ internal/framework/pipeline/chunk_cache.go | 29 ++ .../framework/pipeline/chunk_cache_test.go | 20 ++ .../framework/pipeline/chunk_plan_store.go | 47 +++ 13 files changed, 843 insertions(+), 1 deletion(-) create mode 100644 internal/core/config/chunk_cache_test.go create mode 100644 internal/core/workspace/chunk_plan.go create mode 100644 internal/core/workspace/chunk_plan_test.go create mode 100644 internal/framework/chunkplan/store.go create mode 100644 internal/framework/chunkplan/store_test.go create mode 100644 internal/framework/pipeline/chunk_cache.go create mode 100644 internal/framework/pipeline/chunk_cache_test.go create mode 100644 internal/framework/pipeline/chunk_plan_store.go diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index d74e70c9..2a0fe14b 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -432,7 +432,7 @@ is green. ## Stage 3: Add dedicated cache configuration and the plan store -**Status:** Not started +**Status:** Complete ### Objective diff --git a/internal/core/config/chunk_cache_test.go b/internal/core/config/chunk_cache_test.go new file mode 100644 index 00000000..587e0c2b --- /dev/null +++ b/internal/core/config/chunk_cache_test.go @@ -0,0 +1,126 @@ +package config + +import ( + "path/filepath" + "strings" + "testing" + + "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" +) + +func TestChunkCacheDefaults(t *testing.T) { + cfg := Default() + if cfg.Workspace.ChunkCache.Mode != pipeline.ChunkCacheAuto || cfg.Workspace.ChunkCache.Directory != "" { + t.Fatalf("chunk cache defaults = %#v", cfg.Workspace.ChunkCache) + } + if err := cfg.Validate(); err != nil { + t.Fatalf("Validate() error = %v", err) + } +} + +func TestChunkCacheFileConfiguration(t *testing.T) { + cfg := parseAndApplyConfig(t, ` +version: 2 +workspace: + chunk_cache: + mode: refresh + directory: " ./state/../plans " +`) + if cfg.Workspace.ChunkCache.Mode != pipeline.ChunkCacheRefresh { + t.Fatalf("mode = %q", cfg.Workspace.ChunkCache.Mode) + } + if got, want := cfg.Workspace.ChunkCache.Directory, filepath.Clean("./state/../plans"); got != want { + t.Fatalf("directory = %q, want %q", got, want) + } + if err := cfg.Validate(); err != nil { + t.Fatalf("Validate() error = %v", err) + } +} + +func TestChunkCacheEnvironmentOverridesFile(t *testing.T) { + fileCfg, err := ParseFileConfigYAML([]byte(` +version: 2 +workspace: + chunk_cache: + mode: bypass + directory: /file/plans +`)) + if err != nil { + t.Fatal(err) + } + cfg := Default() + if err := cfg.ApplyFileConfigWithLookup(fileCfg, emptyLookup); err != nil { + t.Fatal(err) + } + if err := cfg.ApplyEnvOverridesWithLookup(mapLookup(map[string]string{ + "NOTARIUS_WORKSPACE_CHUNK_CACHE_MODE": "refresh", + "NOTARIUS_WORKSPACE_CHUNK_CACHE_DIR": " /environment/../cache/plans ", + })); err != nil { + t.Fatal(err) + } + if cfg.Workspace.ChunkCache.Mode != pipeline.ChunkCacheRefresh || cfg.Workspace.ChunkCache.Directory != filepath.Clean("/environment/../cache/plans") { + t.Fatalf("effective chunk cache = %#v", cfg.Workspace.ChunkCache) + } +} + +func TestChunkCacheEmptyDirectoryEnvironmentSelectsDefault(t *testing.T) { + cfg := Default() + cfg.Workspace.ChunkCache.Directory = "/file/plans" + if err := cfg.ApplyEnvOverridesWithLookup(mapLookup(map[string]string{"NOTARIUS_WORKSPACE_CHUNK_CACHE_DIR": " \t "})); err != nil { + t.Fatal(err) + } + if cfg.Workspace.ChunkCache.Directory != "" { + t.Fatalf("directory = %q, want unset", cfg.Workspace.ChunkCache.Directory) + } +} + +func TestChunkCacheRejectsInvalidSuppliedModes(t *testing.T) { + fileCfg, err := ParseFileConfigYAML([]byte("version: 2\nworkspace:\n chunk_cache:\n mode: sometimes\n")) + if err != nil { + t.Fatal(err) + } + cfg := Default() + if err := cfg.ApplyFileConfigWithLookup(fileCfg, emptyLookup); err == nil || !strings.Contains(err.Error(), "workspace.chunk_cache.mode") { + t.Fatalf("file mode error = %v", err) + } + + cfg = Default() + if err := cfg.ApplyEnvOverridesWithLookup(mapLookup(map[string]string{"NOTARIUS_WORKSPACE_CHUNK_CACHE_MODE": "sometimes"})); err == nil || !strings.Contains(err.Error(), "NOTARIUS_WORKSPACE_CHUNK_CACHE_MODE") { + t.Fatalf("environment mode error = %v", err) + } + + cfg = Default() + cfg.Workspace.ChunkCache.Mode = "sometimes" + if err := cfg.Validate(); err == nil || !strings.Contains(err.Error(), "chunk cache") { + t.Fatalf("Validate() error = %v", err) + } +} + +func TestChunkCacheConfigurationClonesAndRedacts(t *testing.T) { + cfg := Default() + cfg.Workspace.ChunkCache = WorkspaceChunkCacheConfig{Mode: pipeline.ChunkCacheRefresh, Directory: "/var/cache/notarius/chunk-plans"} + cloned := cloneConfig(cfg) + redacted := cfg.Redacted() + if cloned.Workspace.ChunkCache != cfg.Workspace.ChunkCache || redacted.Workspace.ChunkCache != cfg.Workspace.ChunkCache { + t.Fatalf("cloned=%#v redacted=%#v", cloned.Workspace.ChunkCache, redacted.Workspace.ChunkCache) + } + redacted.Workspace.ChunkCache.Directory = "/changed" + if cfg.Workspace.ChunkCache.Directory != "/var/cache/notarius/chunk-plans" { + t.Fatal("redacted mutation changed original") + } + if err := cfg.Validate(); err != nil { + t.Fatalf("Validate() error = %v", err) + } +} + +func TestChunkCacheDirectoryValidation(t *testing.T) { + cfg := Default() + cfg.Workspace.ChunkCache.Directory = "/var/cache/notarius/chunk-plans" + if err := cfg.Validate(); err != nil { + t.Fatalf("Validate(system root) error = %v", err) + } + cfg.Workspace.ChunkCache.Directory = "bad\x00path" + if err := cfg.Validate(); err == nil || !strings.Contains(err.Error(), "NUL") { + t.Fatalf("Validate(NUL directory) error = %v", err) + } +} diff --git a/internal/core/config/config.go b/internal/core/config/config.go index e091f851..190f6cd2 100644 --- a/internal/core/config/config.go +++ b/internal/core/config/config.go @@ -38,11 +38,17 @@ type DiagnosticsConfig struct { type WorkspaceConfig struct { Directory string `json:"directory,omitempty"` + ChunkCache WorkspaceChunkCacheConfig `json:"chunk_cache"` Diagnostics WorkspaceDiagnosticsConfig `json:"diagnostics"` Resume WorkspaceResumeConfig `json:"resume"` Debug WorkspaceDebugConfig `json:"debug"` } +type WorkspaceChunkCacheConfig struct { + Mode pipeline.ChunkCacheMode `json:"mode"` + Directory string `json:"directory,omitempty"` +} + type WorkspaceDiagnosticsConfig struct { Enabled bool `json:"enabled"` Retention diagnostics.RetentionMode `json:"retention,omitempty"` @@ -71,6 +77,7 @@ func Default() Config { Retention: diagnostics.RetentionAuto, }, Workspace: WorkspaceConfig{ + ChunkCache: WorkspaceChunkCacheConfig{Mode: pipeline.ChunkCacheAuto}, Diagnostics: WorkspaceDiagnosticsConfig{ Enabled: true, }, diff --git a/internal/core/config/env.go b/internal/core/config/env.go index fb33a28e..82c7231b 100644 --- a/internal/core/config/env.go +++ b/internal/core/config/env.go @@ -7,6 +7,7 @@ import ( "strings" "gitea.maximumdirect.net/eric/notarius/internal/core/diagnostics" + "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" ) func LoadFromEnv() (Config, error) { @@ -57,6 +58,16 @@ func (c *Config) applyEnvOverridesWithLookup(lookup func(string) (string, bool)) if raw, ok := lookup("NOTARIUS_WORKSPACE_DIR"); ok { c.Workspace.Directory = strings.TrimSpace(raw) } + if raw, ok := lookup("NOTARIUS_WORKSPACE_CHUNK_CACHE_MODE"); ok { + mode, err := pipeline.ParseChunkCacheMode(raw) + if err != nil { + return fmt.Errorf("NOTARIUS_WORKSPACE_CHUNK_CACHE_MODE: %w", err) + } + c.Workspace.ChunkCache.Mode = mode + } + if raw, ok := lookup("NOTARIUS_WORKSPACE_CHUNK_CACHE_DIR"); ok { + c.Workspace.ChunkCache.Directory = cleanOptionalPath(raw) + } if raw, ok := lookup("NOTARIUS_WORKSPACE_DIAGNOSTICS_ENABLED"); ok { value, err := parseBoolEnv("NOTARIUS_WORKSPACE_DIAGNOSTICS_ENABLED", raw) if err != nil { diff --git a/internal/core/config/file_config.go b/internal/core/config/file_config.go index dcfa5716..5dacaf47 100644 --- a/internal/core/config/file_config.go +++ b/internal/core/config/file_config.go @@ -4,6 +4,7 @@ import ( "bytes" "fmt" "os" + "path/filepath" "sort" "strings" @@ -54,11 +55,17 @@ type FileDiagnosticsConfig struct { type FileWorkspaceConfig struct { Directory *string `yaml:"directory,omitempty"` + ChunkCache *FileWorkspaceChunkCacheConfig `yaml:"chunk_cache,omitempty"` Diagnostics *FileWorkspaceDiagnosticsConfig `yaml:"diagnostics,omitempty"` Resume *FileWorkspaceEnabledConfig `yaml:"resume,omitempty"` Debug *FileWorkspaceEnabledConfig `yaml:"debug,omitempty"` } +type FileWorkspaceChunkCacheConfig struct { + Mode *string `yaml:"mode,omitempty"` + Directory *string `yaml:"directory,omitempty"` +} + type FileWorkspaceDiagnosticsConfig struct { Enabled *bool `yaml:"enabled,omitempty"` Retention *string `yaml:"retention,omitempty"` @@ -338,6 +345,18 @@ func (c *Config) applyFileConfigWithLookup(fileCfg FileConfig, lookup func(strin if fileCfg.Workspace.Directory != nil { c.Workspace.Directory = strings.TrimSpace(*fileCfg.Workspace.Directory) } + if fileCfg.Workspace.ChunkCache != nil { + if fileCfg.Workspace.ChunkCache.Mode != nil { + mode, err := pipeline.ParseChunkCacheMode(*fileCfg.Workspace.ChunkCache.Mode) + if err != nil { + return fmt.Errorf("workspace.chunk_cache.mode: %w", err) + } + c.Workspace.ChunkCache.Mode = mode + } + if fileCfg.Workspace.ChunkCache.Directory != nil { + c.Workspace.ChunkCache.Directory = cleanOptionalPath(*fileCfg.Workspace.ChunkCache.Directory) + } + } if fileCfg.Workspace.Diagnostics != nil { if fileCfg.Workspace.Diagnostics.Enabled != nil { c.Workspace.Diagnostics.Enabled = *fileCfg.Workspace.Diagnostics.Enabled @@ -360,6 +379,14 @@ func (c *Config) applyFileConfigWithLookup(fileCfg FileConfig, lookup func(strin return nil } +func cleanOptionalPath(value string) string { + value = strings.TrimSpace(value) + if value == "" { + return "" + } + return filepath.Clean(value) +} + func normalizeStageWorkers(values map[string]int) (map[string]int, bool, error) { workers := make(map[string]int, len(values)) configured := false diff --git a/internal/core/config/validation.go b/internal/core/config/validation.go index 7742b1e4..019dc86d 100644 --- a/internal/core/config/validation.go +++ b/internal/core/config/validation.go @@ -61,6 +61,12 @@ func validateScriptorium(cfg ScriptoriumConfig) error { } func validateWorkspace(cfg WorkspaceConfig) error { + if err := cfg.ChunkCache.Mode.Validate(); err != nil { + return fmt.Errorf("workspace chunk cache: %w", err) + } + if strings.ContainsRune(cfg.ChunkCache.Directory, '\x00') { + return fmt.Errorf("workspace chunk cache directory must not contain NUL") + } if cfg.Diagnostics.retentionSet { switch cfg.Diagnostics.Retention { case "", diagnostics.RetentionAuto, diagnostics.RetentionAlways, diagnostics.RetentionNever: diff --git a/internal/core/workspace/chunk_plan.go b/internal/core/workspace/chunk_plan.go new file mode 100644 index 00000000..4fc53426 --- /dev/null +++ b/internal/core/workspace/chunk_plan.go @@ -0,0 +1,22 @@ +package workspace + +import ( + "fmt" + "path/filepath" + "strings" +) + +func DefaultChunkPlanRoot(userCacheDir func() (string, error)) (string, error) { + if userCacheDir == nil { + return "", fmt.Errorf("user cache directory resolver must not be nil") + } + root, err := userCacheDir() + if err != nil { + return "", fmt.Errorf("resolve user cache directory: %w", err) + } + root = strings.TrimSpace(root) + if root == "" { + return "", fmt.Errorf("user cache directory must not be empty") + } + return filepath.Join(filepath.Clean(root), "notarius", "chunk-plans"), nil +} diff --git a/internal/core/workspace/chunk_plan_test.go b/internal/core/workspace/chunk_plan_test.go new file mode 100644 index 00000000..b806848d --- /dev/null +++ b/internal/core/workspace/chunk_plan_test.go @@ -0,0 +1,57 @@ +package workspace + +import ( + "errors" + "path/filepath" + "strings" + "testing" + + "gitea.maximumdirect.net/eric/notarius/internal/core/config" +) + +func TestDefaultChunkPlanRoot(t *testing.T) { + got, err := DefaultChunkPlanRoot(func() (string, error) { return "/cache/user", nil }) + if err != nil { + t.Fatalf("DefaultChunkPlanRoot() error = %v", err) + } + if want := filepath.Join("/cache/user", "notarius", "chunk-plans"); got != want { + t.Fatalf("root = %q, want %q", got, want) + } +} + +func TestDefaultChunkPlanRootRejectsResolverFailures(t *testing.T) { + boom := errors.New("resolver failed") + if _, err := DefaultChunkPlanRoot(func() (string, error) { return "", boom }); !errors.Is(err, boom) { + t.Fatalf("resolver error = %v", err) + } + if _, err := DefaultChunkPlanRoot(func() (string, error) { return " \t ", nil }); err == nil || !strings.Contains(err.Error(), "empty") { + t.Fatalf("empty result error = %v", err) + } + if _, err := DefaultChunkPlanRoot(nil); err == nil { + t.Fatal("nil resolver error = nil") + } +} + +func TestChunkPlanDirectoryIsIndependentFromWorkspaceSettings(t *testing.T) { + base := config.Default() + base.Workspace.Directory = "/workspace/one" + base.Workspace.ChunkCache.Directory = "/cache/plans" + first := FromConfig(base) + + changedWorkspace := base + changedWorkspace.Workspace.Directory = "/workspace/two" + second := FromConfig(changedWorkspace) + if base.Workspace.ChunkCache.Directory != changedWorkspace.Workspace.ChunkCache.Directory { + t.Fatal("workspace directory changed chunk plan directory") + } + if first.RootDir == second.RootDir || first.CheckpointsRoot == second.CheckpointsRoot || first.DebugRoot == second.DebugRoot { + t.Fatalf("workspace settings did not follow workspace directory: %#v %#v", first, second) + } + + changedCache := base + changedCache.Workspace.ChunkCache.Directory = "/cache/other" + third := FromConfig(changedCache) + if first != third { + t.Fatalf("chunk plan directory changed workspace settings: %#v %#v", first, third) + } +} diff --git a/internal/framework/chunkplan/store.go b/internal/framework/chunkplan/store.go new file mode 100644 index 00000000..f076ca80 --- /dev/null +++ b/internal/framework/chunkplan/store.go @@ -0,0 +1,207 @@ +package chunkplan + +import ( + "bytes" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "os" + "path/filepath" + "strings" + + "gitea.maximumdirect.net/eric/notarius/internal/core/source" + "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" +) + +const SchemaVersion = "notarius.chunk-plan.v1" + +type filesystemStore struct { + root string +} + +func NewFilesystemStore(root string) (pipeline.ChunkPlanStore, error) { + root = strings.TrimSpace(root) + if root == "" { + return nil, fmt.Errorf("chunk plan root must not be empty") + } + if strings.ContainsRune(root, '\x00') { + return nil, fmt.Errorf("chunk plan root must not contain NUL") + } + return &filesystemStore{root: filepath.Clean(root)}, nil +} + +func (s *filesystemStore) Load(sourceDigest string) (pipeline.ChunkPlanRecord, pipeline.ChunkPlanDecision, error) { + target, err := s.planPath(sourceDigest) + if err != nil { + return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{}, err + } + data, err := os.ReadFile(target) + if err != nil { + if os.IsNotExist(err) { + return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{Status: pipeline.ChunkPlanMissing, Reason: "chunk plan not found"}, nil + } + return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{}, fmt.Errorf("read chunk plan: %w", err) + } + + var record pipeline.ChunkPlanRecord + decoder := json.NewDecoder(bytes.NewReader(data)) + decoder.DisallowUnknownFields() + decoder.UseNumber() + if err := decoder.Decode(&record); err != nil { + return invalidDecision(fmt.Sprintf("decode stored chunk plan: %v", err)) + } + var trailing any + if err := decoder.Decode(&trailing); err != io.EOF { + if err == nil { + return invalidDecision("stored chunk plan contains trailing JSON") + } + return invalidDecision(fmt.Sprintf("decode stored chunk plan trailer: %v", err)) + } + if err := validateRecord(record, sourceDigest); err != nil { + return invalidDecision(err.Error()) + } + return record, pipeline.ChunkPlanDecision{Status: pipeline.ChunkPlanHit, Reason: "stored chunk plan is valid"}, nil +} + +func (s *filesystemStore) Save(record pipeline.ChunkPlanRecord) error { + if err := validateRecord(record, record.SourceDigest); err != nil { + return fmt.Errorf("validate chunk plan record: %w", err) + } + target, err := s.planPath(record.SourceDigest) + if err != nil { + return err + } + data, err := json.Marshal(record) + if err != nil { + return fmt.Errorf("encode chunk plan record: %w", err) + } + data = append(data, '\n') + if err := writeAtomic(target, data); err != nil { + return fmt.Errorf("write chunk plan: %w", err) + } + return nil +} + +func (s *filesystemStore) planPath(sourceDigest string) (string, error) { + if s == nil || strings.TrimSpace(s.root) == "" { + return "", fmt.Errorf("chunk plan store must not be nil") + } + hexDigest, err := digestPathSegment(sourceDigest) + if err != nil { + return "", err + } + return filepath.Join(s.root, hexDigest, "plan.json"), nil +} + +func digestPathSegment(digest string) (string, error) { + if !strings.HasPrefix(digest, "sha256:") { + return "", fmt.Errorf("source digest must use sha256:<64 lowercase hex> format") + } + hexDigest := strings.TrimPrefix(digest, "sha256:") + if len(hexDigest) != 64 || strings.ToLower(hexDigest) != hexDigest { + return "", fmt.Errorf("source digest must use sha256:<64 lowercase hex> format") + } + decoded, err := hex.DecodeString(hexDigest) + if err != nil || len(decoded) != 32 { + return "", fmt.Errorf("source digest must use sha256:<64 lowercase hex> format") + } + return hexDigest, nil +} + +func validateRecord(record pipeline.ChunkPlanRecord, requestedDigest string) error { + if record.SchemaVersion != SchemaVersion { + return fmt.Errorf("schema_version %q is not supported", record.SchemaVersion) + } + if _, err := digestPathSegment(requestedDigest); err != nil { + return err + } + if record.SourceDigest != requestedDigest { + return fmt.Errorf("source_digest does not match requested source") + } + if record.Plan.SourceDigest != record.SourceDigest { + return fmt.Errorf("plan source_digest does not match record source_digest") + } + if len(record.Plan.Ranges) == 0 { + return fmt.Errorf("plan ranges must not be empty") + } + if err := source.ValidateChunkAnnotations(record.Plan.Annotations); err != nil { + return fmt.Errorf("plan annotations: %w", err) + } + seenStarts := make(map[int]struct{}, len(record.Plan.Ranges)) + for i, chunkRange := range record.Plan.Ranges { + if chunkRange.StartUnitID <= 0 || chunkRange.EndUnitID <= 0 { + return fmt.Errorf("plan range[%d] boundaries must be positive", i) + } + if _, exists := seenStarts[chunkRange.StartUnitID]; exists { + return fmt.Errorf("plan range[%d] duplicates start_unit_id %d", i, chunkRange.StartUnitID) + } + seenStarts[chunkRange.StartUnitID] = struct{}{} + if err := source.ValidateChunkAnnotations(chunkRange.Annotations); err != nil { + return fmt.Errorf("plan range[%d] annotations: %w", i, err) + } + } + wantDigest, err := source.DigestChunkPlan(record.Plan) + if err != nil { + return fmt.Errorf("digest plan: %w", err) + } + if record.PlanDigest != wantDigest { + return fmt.Errorf("plan_digest does not match plan") + } + if strings.TrimSpace(record.Producer.InputModule) == "" { + return fmt.Errorf("producer input_module must not be empty") + } + if strings.TrimSpace(record.Producer.ChunkModule) == "" { + return fmt.Errorf("producer chunk_module must not be empty") + } + if record.CreatedAt.IsZero() { + return fmt.Errorf("created_at must not be zero") + } + return nil +} + +func invalidDecision(reason string) (pipeline.ChunkPlanRecord, pipeline.ChunkPlanDecision, error) { + return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{Status: pipeline.ChunkPlanInvalid, Reason: reason}, nil +} + +func writeAtomic(target string, data []byte) error { + dir := filepath.Dir(target) + if err := os.MkdirAll(dir, 0o700); err != nil { + return err + } + if err := os.Chmod(dir, 0o700); err != nil { + return err + } + + temp, err := os.CreateTemp(dir, ".plan.json.tmp-*") + if err != nil { + return err + } + tempPath := temp.Name() + removeTemp := true + defer func() { + if removeTemp { + _ = os.Remove(tempPath) + } + }() + if err := temp.Chmod(0o600); err != nil { + _ = temp.Close() + return err + } + if _, err := temp.Write(data); err != nil { + _ = temp.Close() + return err + } + if err := temp.Sync(); err != nil { + _ = temp.Close() + return err + } + if err := temp.Close(); err != nil { + return err + } + if err := os.Rename(tempPath, target); err != nil { + return err + } + removeTemp = false + return nil +} diff --git a/internal/framework/chunkplan/store_test.go b/internal/framework/chunkplan/store_test.go new file mode 100644 index 00000000..a4e46461 --- /dev/null +++ b/internal/framework/chunkplan/store_test.go @@ -0,0 +1,283 @@ +package chunkplan + +import ( + "bytes" + "encoding/json" + "os" + "path/filepath" + "reflect" + "strings" + "sync" + "testing" + "time" + + "gitea.maximumdirect.net/eric/notarius/internal/core/artifacts" + "gitea.maximumdirect.net/eric/notarius/internal/core/source" + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" + "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" +) + +const testSourceDigest = "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + +func TestFilesystemStoreRoundTripAndExactPath(t *testing.T) { + root := filepath.Join(t.TempDir(), "plans") + store := newStore(t, root) + record := testRecord(t, 1) + if err := store.Save(record); err != nil { + t.Fatalf("Save() error = %v", err) + } + + target := filepath.Join(root, strings.TrimPrefix(testSourceDigest, "sha256:"), "plan.json") + data, err := os.ReadFile(target) + if err != nil { + t.Fatalf("read exact plan path: %v", err) + } + if bytes.Contains(data, []byte("RAW REFERENCE CONTENT")) { + t.Fatal("stored record contains raw reference content") + } + var envelope struct { + Producer struct { + References []map[string]any `json:"references"` + } `json:"producer"` + } + if err := json.Unmarshal(data, &envelope); err != nil { + t.Fatal(err) + } + if len(envelope.Producer.References) != 1 { + t.Fatalf("stored references = %#v", envelope.Producer.References) + } + if _, exists := envelope.Producer.References[0]["content"]; exists { + t.Fatalf("stored reference contains content field: %#v", envelope.Producer.References[0]) + } + + got, decision, err := store.Load(testSourceDigest) + if err != nil || decision.Status != pipeline.ChunkPlanHit { + t.Fatalf("Load() decision=%#v error=%v", decision, err) + } + if !reflect.DeepEqual(got, record) { + t.Fatalf("round trip record = %#v, want %#v", got, record) + } +} + +func TestFilesystemStorePermissions(t *testing.T) { + root := filepath.Join(t.TempDir(), "plans") + store := newStore(t, root) + if err := store.Save(testRecord(t, 1)); err != nil { + t.Fatal(err) + } + digestDir := filepath.Join(root, strings.TrimPrefix(testSourceDigest, "sha256:")) + for path, want := range map[string]os.FileMode{ + root: 0o700, + digestDir: 0o700, + filepath.Join(digestDir, "plan.json"): 0o600, + } { + info, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + if got := info.Mode().Perm(); got != want { + t.Fatalf("%s mode = %04o, want %04o", path, got, want) + } + } +} + +func TestFilesystemStoreMissingAndOperationalErrors(t *testing.T) { + root := t.TempDir() + store := newStore(t, root) + _, decision, err := store.Load(testSourceDigest) + if err != nil || decision.Status != pipeline.ChunkPlanMissing { + t.Fatalf("missing decision=%#v error=%v", decision, err) + } + + target := filepath.Join(root, strings.TrimPrefix(testSourceDigest, "sha256:"), "plan.json") + if err := os.MkdirAll(target, 0o700); err != nil { + t.Fatal(err) + } + if _, _, err := store.Load(testSourceDigest); err == nil { + t.Fatal("Load(plan.json directory) error = nil") + } +} + +func TestFilesystemStoreRejectsMalformedSourceDigests(t *testing.T) { + store := newStore(t, t.TempDir()) + for _, digest := range []string{"", "sha1:" + strings.Repeat("a", 64), "sha256:../escape", "sha256:" + strings.Repeat("A", 64), "sha256:" + strings.Repeat("a", 63)} { + t.Run(digest, func(t *testing.T) { + if _, _, err := store.Load(digest); err == nil { + t.Fatal("Load() error = nil") + } + record := testRecord(t, 1) + record.SourceDigest = digest + record.Plan.SourceDigest = digest + record.PlanDigest, _ = source.DigestChunkPlan(record.Plan) + if err := store.Save(record); err == nil { + t.Fatal("Save() error = nil") + } + }) + } +} + +func TestFilesystemStoreReportsInvalidRecordsAsRecoverable(t *testing.T) { + tests := []struct { + name string + mutate func([]byte) []byte + want string + }{ + {name: "unknown field", mutate: func(data []byte) []byte { + return bytes.Replace(data, []byte(`{"schema_version"`), []byte(`{"unknown":true,"schema_version"`), 1) + }, want: "unknown"}, + {name: "schema mismatch", mutate: replaceJSON(`notarius.chunk-plan.v1`, `notarius.chunk-plan.v2`), want: "schema_version"}, + {name: "source mismatch", mutate: replaceJSON(testSourceDigest, "sha256:"+strings.Repeat("b", 64)), want: "source_digest"}, + {name: "plan digest mismatch", mutate: func(data []byte) []byte { + prefix := []byte(`"plan_digest":"sha256:`) + index := bytes.Index(data, prefix) + if index >= 0 { + data[index+len(prefix)] = '0' + } + return data + }, want: "plan_digest"}, + {name: "noncanonical annotation", mutate: func(data []byte) []byte { + return bytes.Replace(data, []byte(`{"value":1}`), []byte(`{ "value": 1 }`), 1) + }, want: "canonical JSON"}, + {name: "trailing JSON", mutate: func(data []byte) []byte { return append(data, []byte(` {}`)...) }, want: "trailing"}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + root := t.TempDir() + store := newStore(t, root) + record := testRecord(t, 1) + if err := store.Save(record); err != nil { + t.Fatal(err) + } + path := filepath.Join(root, strings.TrimPrefix(testSourceDigest, "sha256:"), "plan.json") + data, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, tc.mutate(data), 0o600); err != nil { + t.Fatal(err) + } + got, decision, err := store.Load(testSourceDigest) + if err != nil || decision.Status != pipeline.ChunkPlanInvalid || !reflect.DeepEqual(got, pipeline.ChunkPlanRecord{}) || !strings.Contains(decision.Reason, tc.want) { + t.Fatalf("record=%#v decision=%#v error=%v", got, decision, err) + } + }) + } +} + +func TestFilesystemStoreAtomicallyReplacesAndPreservesValidRecordOnFailure(t *testing.T) { + root := t.TempDir() + store := newStore(t, root) + first := testRecord(t, 1) + second := testRecord(t, 2) + if err := store.Save(first); err != nil { + t.Fatal(err) + } + invalid := second + invalid.PlanDigest = "sha256:" + strings.Repeat("0", 64) + if err := store.Save(invalid); err == nil { + t.Fatal("Save(invalid) error = nil") + } + got, _, _ := store.Load(testSourceDigest) + if !reflect.DeepEqual(got, first) { + t.Fatalf("record after failed replacement = %#v", got) + } + if err := store.Save(second); err != nil { + t.Fatal(err) + } + got, _, _ = store.Load(testSourceDigest) + if !reflect.DeepEqual(got, second) { + t.Fatalf("record after replacement = %#v", got) + } + assertNoTemps(t, filepath.Join(root, strings.TrimPrefix(testSourceDigest, "sha256:"))) +} + +func TestFilesystemStoreConcurrentWritersExposeCompleteRecord(t *testing.T) { + store := newStore(t, t.TempDir()) + const writers = 24 + records := make([]pipeline.ChunkPlanRecord, writers) + for i := range records { + records[i] = testRecord(t, i+1) + } + var wg sync.WaitGroup + errs := make(chan error, writers) + for i := range records { + wg.Add(1) + go func(record pipeline.ChunkPlanRecord) { + defer wg.Done() + errs <- store.Save(record) + }(records[i]) + } + wg.Wait() + close(errs) + for err := range errs { + if err != nil { + t.Fatalf("concurrent Save() error = %v", err) + } + } + got, decision, err := store.Load(testSourceDigest) + if err != nil || decision.Status != pipeline.ChunkPlanHit { + t.Fatalf("Load() decision=%#v error=%v", decision, err) + } + var annotation struct { + Value int `json:"value"` + } + if err := json.Unmarshal(got.Plan.Annotations["test/value"], &annotation); err != nil || annotation.Value < 1 || annotation.Value > writers { + t.Fatalf("final annotation=%#v error=%v", annotation, err) + } +} + +func newStore(t *testing.T, root string) pipeline.ChunkPlanStore { + t.Helper() + store, err := NewFilesystemStore(root) + if err != nil { + t.Fatalf("NewFilesystemStore() error = %v", err) + } + return store +} + +func testRecord(t *testing.T, value int) pipeline.ChunkPlanRecord { + t.Helper() + annotation, err := json.Marshal(map[string]int{"value": value}) + if err != nil { + t.Fatal(err) + } + plan := source.ChunkPlan{ + SourceDigest: testSourceDigest, + Ranges: []source.ChunkRange{{StartUnitID: 1, EndUnitID: 2, Annotations: source.ChunkAnnotations{"test/range": json.RawMessage(`{"range":true}`)}}}, + Annotations: source.ChunkAnnotations{"test/value": annotation}, + } + planDigest, err := source.DigestChunkPlan(plan) + if err != nil { + t.Fatal(err) + } + return pipeline.ChunkPlanRecord{ + SchemaVersion: SchemaVersion, + SourceDigest: testSourceDigest, + PlanDigest: planDigest, + Plan: plan, + Producer: pipeline.ChunkPlanProducer{ + InputModule: "input/test", ChunkModule: "chunk/test", LLMProfile: "profile/test", + References: []artifacts.ReferenceProvenance{{Stage: "chunk", SlotName: "guide", OriginType: "file", OriginURI: "file:///guide.txt", Digest: "sha256:reference"}}, + Metadata: map[string]any{"prompt_id": "test/prompt", "enabled": true}, + }, + Warnings: []contracts.Warning{{Scope: "chunk/test", ReasonCode: "observed", Message: "warning"}}, + CreatedAt: time.Date(2026, 7, 18, 12, 0, 0, 0, time.UTC), + } +} + +func replaceJSON(old, replacement string) func([]byte) []byte { + return func(data []byte) []byte { return bytes.Replace(data, []byte(old), []byte(replacement), 1) } +} + +func assertNoTemps(t *testing.T, dir string) { + t.Helper() + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatal(err) + } + for _, entry := range entries { + if strings.Contains(entry.Name(), ".tmp-") { + t.Fatalf("temporary file remains: %s", entry.Name()) + } + } +} diff --git a/internal/framework/pipeline/chunk_cache.go b/internal/framework/pipeline/chunk_cache.go new file mode 100644 index 00000000..ff0b8dca --- /dev/null +++ b/internal/framework/pipeline/chunk_cache.go @@ -0,0 +1,29 @@ +package pipeline + +import ( + "fmt" + "strings" +) + +type ChunkCacheMode string + +const ( + ChunkCacheAuto ChunkCacheMode = "auto" + ChunkCacheBypass ChunkCacheMode = "bypass" + ChunkCacheRefresh ChunkCacheMode = "refresh" +) + +func ParseChunkCacheMode(raw string) (ChunkCacheMode, error) { + mode := ChunkCacheMode(strings.TrimSpace(raw)) + switch mode { + case ChunkCacheAuto, ChunkCacheBypass, ChunkCacheRefresh: + return mode, nil + default: + return "", fmt.Errorf("chunk cache mode %q is not supported", strings.TrimSpace(raw)) + } +} + +func (m ChunkCacheMode) Validate() error { + _, err := ParseChunkCacheMode(string(m)) + return err +} diff --git a/internal/framework/pipeline/chunk_cache_test.go b/internal/framework/pipeline/chunk_cache_test.go new file mode 100644 index 00000000..11334e6d --- /dev/null +++ b/internal/framework/pipeline/chunk_cache_test.go @@ -0,0 +1,20 @@ +package pipeline + +import "testing" + +func TestParseChunkCacheMode(t *testing.T) { + for _, value := range []ChunkCacheMode{ChunkCacheAuto, ChunkCacheBypass, ChunkCacheRefresh} { + got, err := ParseChunkCacheMode(string(value)) + if err != nil || got != value { + t.Fatalf("ParseChunkCacheMode(%q) = %q, %v", value, got, err) + } + if err := value.Validate(); err != nil { + t.Fatalf("%q.Validate() error = %v", value, err) + } + } + for _, value := range []string{"", "enabled", "AUTO", "auto,refresh"} { + if _, err := ParseChunkCacheMode(value); err == nil { + t.Fatalf("ParseChunkCacheMode(%q) error = nil, want error", value) + } + } +} diff --git a/internal/framework/pipeline/chunk_plan_store.go b/internal/framework/pipeline/chunk_plan_store.go new file mode 100644 index 00000000..e33efb16 --- /dev/null +++ b/internal/framework/pipeline/chunk_plan_store.go @@ -0,0 +1,47 @@ +package pipeline + +import ( + "time" + + "gitea.maximumdirect.net/eric/notarius/internal/core/artifacts" + "gitea.maximumdirect.net/eric/notarius/internal/core/source" + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" +) + +type ChunkPlanProducer struct { + InputModule string `json:"input_module"` + ChunkModule string `json:"chunk_module"` + LLMProfile string `json:"llm_profile,omitempty"` + References []artifacts.ReferenceProvenance `json:"references,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` +} + +type ChunkPlanRecord struct { + SchemaVersion string `json:"schema_version"` + SourceDigest string `json:"source_digest"` + PlanDigest string `json:"plan_digest"` + Plan source.ChunkPlan `json:"plan"` + Producer ChunkPlanProducer `json:"producer"` + Warnings []contracts.Warning `json:"warnings,omitempty"` + CreatedAt time.Time `json:"created_at"` +} + +type ChunkPlanStore interface { + Load(sourceDigest string) (ChunkPlanRecord, ChunkPlanDecision, error) + Save(ChunkPlanRecord) error +} + +type ChunkPlanStoreFactory func(root string) (ChunkPlanStore, error) + +type ChunkPlanDecision struct { + Status ChunkPlanStatus `json:"status"` + Reason string `json:"reason,omitempty"` +} + +type ChunkPlanStatus string + +const ( + ChunkPlanHit ChunkPlanStatus = "hit" + ChunkPlanMissing ChunkPlanStatus = "missing" + ChunkPlanInvalid ChunkPlanStatus = "invalid" +)