Compare commits
5 Commits
b3328b93e5
...
604c7a7945
| Author | SHA1 | Date | |
|---|---|---|---|
| 604c7a7945 | |||
| a0f5e6e2b9 | |||
| 561d65a505 | |||
| 205e2a9908 | |||
| 8c59b6af14 |
@@ -32,7 +32,10 @@ digests, requested module, lookup decision, materialization action, validation
|
||||
decision, and publication decision. Its closed decision values make failures
|
||||
and recoverable invalid records inspectable without serializing plan ranges,
|
||||
annotations, source content, reference content, prompts, model responses, or
|
||||
raw invalid-file bytes.
|
||||
raw invalid-file bytes. Lookup reasons are derived only from the lookup status:
|
||||
`stored chunk plan is valid`, `chunk plan not found`, `stored chunk plan is
|
||||
invalid`, or `chunk plan lookup skipped`. Store-provided reasons and malformed
|
||||
record details never enter this artifact.
|
||||
|
||||
JSON methods indent their payload and append a newline. All artifact writes use
|
||||
a temporary file in the target directory, apply the requested permissions, and
|
||||
|
||||
@@ -71,7 +71,8 @@ Chunkers implement `contracts.Chunker.Plan`. A plan identifies ordered source
|
||||
unit ranges and may carry optional namespaced JSON annotations; it does not
|
||||
contain materialized chunk content. The framework canonicalizes annotations,
|
||||
validates ranges against the current source, and materializes chunk IDs,
|
||||
indexes, references, content, units, and generic metadata. Annotation
|
||||
indexes, references, content, units, and generic metadata. Materialized source
|
||||
unit metadata is independently owned. Annotation
|
||||
namespaces remain optional data: generic framework code and downstream modules
|
||||
must not require D&D scene annotations or import `dnd/scenes`.
|
||||
|
||||
|
||||
@@ -34,7 +34,7 @@ normalize continuations that may overlap across lanes.
|
||||
| `internal/core/artifacts` | Run-manifest and provenance models. |
|
||||
| `internal/core/config` | Defaults, YAML parsing, environment overrides, validation, redaction, and effective pipeline resolution. |
|
||||
| `internal/core/diagnostics` | Scoped run directories, diagnostics writers, atomic writes, and retention decisions. |
|
||||
| `internal/core/source` | Generic source documents, units, chunks, canonical references, lookup, validation, and deterministic source digests. |
|
||||
| `internal/core/source` | Generic source documents, units, chunks, canonical references, validation, deterministic source digests, and independent metadata materialization. |
|
||||
| `internal/core/workspace` | Effective workspace settings, confined paths and writes, and checkpoint identity and manifest models. |
|
||||
|
||||
## Framework Packages
|
||||
|
||||
@@ -214,7 +214,10 @@ Plan canonicalization requires canonical JSON annotations, a matching source
|
||||
digest, at least one range, existing ordered boundaries, and increasing range
|
||||
starts. Ranges may overlap or leave gaps; a chunker may impose stricter policy.
|
||||
Materialization deterministically reconstructs each range from the current
|
||||
source document and copies annotations without interpreting their namespaces.
|
||||
source document, deep-clones JSON-shaped source-unit metadata, and copies
|
||||
annotations without interpreting their namespaces. Materialized chunks and
|
||||
separate materializations do not share mutable unit metadata; unsupported or
|
||||
cyclic metadata fails materialization with context.
|
||||
|
||||
Before lane execution, generic chunk validation checks the materialized chunks'
|
||||
identities, order, source references, content, media type, units, and metadata.
|
||||
|
||||
@@ -97,6 +97,11 @@ an `auto` run regenerates and atomically replaces it only after validation
|
||||
succeeds. Delete an exact cache root or digest directory only when regeneration
|
||||
cost is acceptable.
|
||||
|
||||
The configured root is the cache trust boundary. An operator-supplied root path
|
||||
may itself resolve through a symlink, but cache-owned digest directories and
|
||||
plan files must be real directory and regular-file entries. Links or other
|
||||
unexpected entry types are rejected rather than followed.
|
||||
|
||||
Publication uses atomic replacement. Concurrent readers observe a complete old
|
||||
or new plan, and concurrent writers leave one complete valid winner; there is
|
||||
no history, lock protocol, or rollback facility. Do not share a cache root
|
||||
|
||||
@@ -1,9 +1,8 @@
|
||||
# ADR-0005 Feature Roadmap
|
||||
|
||||
This roadmap defines the intended end state for
|
||||
[ADR-0005](../adr/0005-cache-canonical-chunk-plans-by-source.md). The feature
|
||||
is not implemented. The ordered coding work belongs in the
|
||||
[implementation plan](implementation.md).
|
||||
This roadmap records the implemented target state for
|
||||
[ADR-0005](../adr/0005-cache-canonical-chunk-plans-by-source.md). The
|
||||
[implementation record](implementation.md) preserves the completed work.
|
||||
|
||||
## Intent
|
||||
|
||||
@@ -156,11 +155,11 @@ This feature does not provide:
|
||||
|
||||
## Completion Outcomes
|
||||
|
||||
The feature is complete when independent runs over the same canonical source
|
||||
**Achieved.** Independent runs over the same canonical source
|
||||
reuse byte-stable materialized chunks across pipeline, lane, reference,
|
||||
chunk-module, and LLM configuration changes; bypass and refresh obey their
|
||||
documented state semantics; provenance identifies the effective producer;
|
||||
legacy chunk checkpoints cannot compete with plan reuse; and all focused,
|
||||
integration, compatibility, CLI, and repository-wide validation passes. The
|
||||
integration, compatibility, CLI, and repository-wide validation pass. The
|
||||
configuration and operations references must also document both the per-user
|
||||
default and the recommended system-wide Linux setting.
|
||||
|
||||
@@ -4,11 +4,9 @@ This document records the implementation of the target state in
|
||||
[ADR-0005 Feature Roadmap](adr0005.md), governed by
|
||||
[ADR-0005](../adr/0005-cache-canonical-chunk-plans-by-source.md).
|
||||
|
||||
Stages 1 through 8 summarize the completed initial implementation. Stages 9
|
||||
through 12 track pending remediation identified by the post-implementation
|
||||
review. Current behavior is documented in the canonical references linked from
|
||||
[Development](../development.md); those references must not describe the Stages
|
||||
9 through 12 target state until the corresponding stage is complete.
|
||||
All stages are complete. This document remains a concise historical
|
||||
implementation record. Current behavior is documented in the canonical
|
||||
references linked from [Development](../development.md).
|
||||
|
||||
## Execution Rules
|
||||
|
||||
@@ -351,7 +349,7 @@ feature and its per-user and system-wide deployment guidance.
|
||||
|
||||
## Stage 9: Redact invalid cache-record diagnostics
|
||||
|
||||
**Status:** Pending
|
||||
**Status:** Complete
|
||||
|
||||
### Objective
|
||||
|
||||
@@ -408,7 +406,7 @@ remain correctly classified, and all focused and repository-wide checks pass.
|
||||
|
||||
## Stage 10: Enforce rooted chunk-plan filesystem access
|
||||
|
||||
**Status:** Pending
|
||||
**Status:** Complete
|
||||
|
||||
### Objective
|
||||
|
||||
@@ -469,7 +467,7 @@ concurrency tests and the race detector pass, and the repository is green.
|
||||
|
||||
## Stage 11: Deep-clone source metadata during materialization
|
||||
|
||||
**Status:** Pending
|
||||
**Status:** Complete
|
||||
|
||||
### Objective
|
||||
|
||||
@@ -526,7 +524,7 @@ current materialized bytes remain stable; and all checks pass.
|
||||
|
||||
## Stage 12: Finalize remediation documentation and roadmap status
|
||||
|
||||
**Status:** Pending
|
||||
**Status:** Complete
|
||||
|
||||
### Objective
|
||||
|
||||
|
||||
@@ -255,6 +255,22 @@ func TestRunRecordsExplicitChunkCacheOverrideAndEffectiveMode(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunChunkPlanDiagnosticsRedactStoreDecisionReason(t *testing.T) {
|
||||
diagnosticsDir := t.TempDir()
|
||||
configPath := writeTestConfig(t, cacheTestConfig("auto", t.TempDir(), diagnosticsDir))
|
||||
factory := &recordingChunkPlanFactory{store: &recordingChunkPlanStore{
|
||||
decision: pipeline.ChunkPlanDecision{Status: pipeline.ChunkPlanInvalid, Reason: "SENTINEL_INVALID_RECORD_CONTENT"},
|
||||
}}
|
||||
code, stderr := runCacheCommand(t, cacheRunArgs(t, configPath), cacheTestOptions(t, nil, factory))
|
||||
if code != 0 || stderr != "" {
|
||||
t.Fatalf("code=%d stderr=%q", code, stderr)
|
||||
}
|
||||
summary := string(readFile(t, filepath.Join(onlyChildDir(t, diagnosticsDir), diagnostics.ArtifactChunkPlan)))
|
||||
if strings.Contains(summary, "SENTINEL_INVALID_RECORD_CONTENT") || !strings.Contains(summary, `"lookup_reason": "stored chunk plan is invalid"`) {
|
||||
t.Fatalf("chunk plan diagnostic = %s", summary)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDefaultAutoReusesPlanAcrossIndependentInvocations(t *testing.T) {
|
||||
cacheBase := filepath.Join(t.TempDir(), "cache")
|
||||
workspaceDir := filepath.Join(t.TempDir(), "workspace")
|
||||
|
||||
@@ -162,7 +162,10 @@ func MaterializeChunkPlan(doc *SourceDocument, plan ChunkPlan) ([]Chunk, error)
|
||||
for index, chunkRange := range plan.Ranges {
|
||||
start, _ := UnitIndex(doc, chunkRange.StartUnitID)
|
||||
end, _ := UnitIndex(doc, chunkRange.EndUnitID)
|
||||
units := cloneSourceUnits(doc.Units[start : end+1])
|
||||
units, err := cloneSourceUnits(doc.Units[start : end+1])
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("clone chunk plan range[%d] units: %w", index, err)
|
||||
}
|
||||
content, err := json.Marshal(struct {
|
||||
Units []SourceUnit `json:"units"`
|
||||
}{Units: units})
|
||||
@@ -189,52 +192,18 @@ func MaterializeChunkPlan(doc *SourceDocument, plan ChunkPlan) ([]Chunk, error)
|
||||
return chunks, nil
|
||||
}
|
||||
|
||||
func cloneSourceUnits(units []SourceUnit) []SourceUnit {
|
||||
func cloneSourceUnits(units []SourceUnit) ([]SourceUnit, error) {
|
||||
if len(units) == 0 {
|
||||
return nil
|
||||
return nil, nil
|
||||
}
|
||||
cloned := make([]SourceUnit, len(units))
|
||||
for i, unit := range units {
|
||||
cloned[i] = unit
|
||||
cloned[i].Metadata = cloneJSONMap(unit.Metadata)
|
||||
}
|
||||
return cloned
|
||||
}
|
||||
|
||||
func cloneJSONMap(values map[string]any) map[string]any {
|
||||
if len(values) == 0 {
|
||||
return nil
|
||||
}
|
||||
cloned := make(map[string]any, len(values))
|
||||
for key, value := range values {
|
||||
cloned[key] = cloneJSONValue(value)
|
||||
}
|
||||
return cloned
|
||||
}
|
||||
|
||||
func cloneJSONValue(value any) any {
|
||||
switch typed := value.(type) {
|
||||
case map[string]any:
|
||||
return cloneJSONMap(typed)
|
||||
case []any:
|
||||
cloned := make([]any, len(typed))
|
||||
for i := range typed {
|
||||
cloned[i] = cloneJSONValue(typed[i])
|
||||
}
|
||||
return cloned
|
||||
case json.RawMessage:
|
||||
return append(json.RawMessage(nil), typed...)
|
||||
case []byte:
|
||||
return append([]byte(nil), typed...)
|
||||
case []string:
|
||||
return append([]string(nil), typed...)
|
||||
case map[string]string:
|
||||
cloned := make(map[string]string, len(typed))
|
||||
for key, item := range typed {
|
||||
cloned[key] = item
|
||||
}
|
||||
return cloned
|
||||
default:
|
||||
return value
|
||||
metadata, err := CloneMetadata(unit.Metadata)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("source unit[%d] metadata: %w", i, err)
|
||||
}
|
||||
cloned[i].Metadata = metadata
|
||||
}
|
||||
return cloned, nil
|
||||
}
|
||||
|
||||
@@ -3,11 +3,16 @@ package source
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"math"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
type typedMetadataMap map[string]any
|
||||
type typedMetadataSlice []typedMetadataMap
|
||||
type typedMetadataArray [2]any
|
||||
|
||||
func TestCanonicalizeChunkAnnotations(t *testing.T) {
|
||||
original := ChunkAnnotations{
|
||||
"domain/items": json.RawMessage(` { "z": [3, 2, 1], "a": 1.0 } `),
|
||||
@@ -218,6 +223,87 @@ func TestMaterializeChunkPlanDeepClonesUnitMetadata(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestMaterializeChunkPlanClonesConcreteJSONMetadata(t *testing.T) {
|
||||
doc := planDocument()
|
||||
doc.Units[0].Metadata = map[string]any{
|
||||
"typed_map": typedMetadataMap{"bytes": []byte("map")},
|
||||
"typed_slice": typedMetadataSlice{{"raw": json.RawMessage(`{"slice":true}`)}},
|
||||
"typed_array": typedMetadataArray{map[string]any{"bytes": []byte("array")}, []any{json.RawMessage(`{"array":true}`)}},
|
||||
"interface": any(typedMetadataMap{"bytes": []byte("interface")}),
|
||||
"raw": json.RawMessage(`{"raw":true}`),
|
||||
"bytes": []byte("bytes"),
|
||||
}
|
||||
|
||||
chunks, err := MaterializeChunkPlan(doc, validChunkPlan(doc))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
again, err := MaterializeChunkPlan(doc, validChunkPlan(doc))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
metadata := chunks[0].Units[0].Metadata
|
||||
metadata["typed_map"].(typedMetadataMap)["bytes"].([]byte)[0] = 'M'
|
||||
metadata["typed_slice"].(typedMetadataSlice)[0]["raw"].(json.RawMessage)[0] = '['
|
||||
metadata["typed_array"].(typedMetadataArray)[0].(map[string]any)["bytes"].([]byte)[0] = 'A'
|
||||
metadata["typed_array"].(typedMetadataArray)[1].([]any)[0].(json.RawMessage)[0] = '['
|
||||
metadata["interface"].(typedMetadataMap)["bytes"].([]byte)[0] = 'I'
|
||||
metadata["raw"].(json.RawMessage)[0] = '['
|
||||
metadata["bytes"].([]byte)[0] = 'B'
|
||||
|
||||
for name, candidate := range map[string]map[string]any{
|
||||
"source": doc.Units[0].Metadata,
|
||||
"again": again[0].Units[0].Metadata,
|
||||
} {
|
||||
if got := string(candidate["typed_map"].(typedMetadataMap)["bytes"].([]byte)); got != "map" {
|
||||
t.Fatalf("%s typed map bytes = %q, want map", name, got)
|
||||
}
|
||||
if got := string(candidate["typed_slice"].(typedMetadataSlice)[0]["raw"].(json.RawMessage)); got != `{"slice":true}` {
|
||||
t.Fatalf("%s typed slice raw = %q", name, got)
|
||||
}
|
||||
array := candidate["typed_array"].(typedMetadataArray)
|
||||
if got := string(array[0].(map[string]any)["bytes"].([]byte)); got != "array" || string(array[1].([]any)[0].(json.RawMessage)) != `{"array":true}` {
|
||||
t.Fatalf("%s typed array = %#v", name, array)
|
||||
}
|
||||
if got := string(candidate["interface"].(typedMetadataMap)["bytes"].([]byte)); got != "interface" {
|
||||
t.Fatalf("%s interface bytes = %q, want interface", name, got)
|
||||
}
|
||||
if got := string(candidate["raw"].(json.RawMessage)); got != `{"raw":true}` {
|
||||
t.Fatalf("%s raw = %q", name, got)
|
||||
}
|
||||
if got := string(candidate["bytes"].([]byte)); got != "bytes" {
|
||||
t.Fatalf("%s bytes = %q, want bytes", name, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestMaterializeChunkPlanRejectsInvalidMetadata(t *testing.T) {
|
||||
cyclic := make(map[string]any)
|
||||
cyclic["self"] = cyclic
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
value any
|
||||
want string
|
||||
}{
|
||||
{name: "cycle", value: cyclic, want: "metadata.cycle.self contains a cycle"},
|
||||
{name: "unsupported", value: func() {}, want: "metadata.unsupported has unsupported type func()"},
|
||||
{name: "nonfinite", value: math.NaN(), want: "metadata.nonfinite has a non-finite number"},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
doc := planDocument()
|
||||
doc.Units[0].Metadata = map[string]any{tc.name: tc.value}
|
||||
_, first := MaterializeChunkPlan(doc, validChunkPlan(doc))
|
||||
_, second := MaterializeChunkPlan(doc, validChunkPlan(doc))
|
||||
if first == nil || !strings.Contains(first.Error(), "clone chunk plan range[0] units: source unit[0] metadata: "+tc.want) {
|
||||
t.Fatalf("first MaterializeChunkPlan() error = %v, want %q", first, tc.want)
|
||||
}
|
||||
if second == nil || second.Error() != first.Error() {
|
||||
t.Fatalf("MaterializeChunkPlan() errors = %v and %v, want deterministic error", first, second)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func planDocument() *SourceDocument {
|
||||
doc := &SourceDocument{ID: "source-plan", Kind: "test", Format: "application/test", Digest: "sha256:source-plan"}
|
||||
for _, id := range []int{10, 20, 30, 40, 50} {
|
||||
|
||||
119
internal/core/source/metadata.go
Normal file
119
internal/core/source/metadata.go
Normal file
@@ -0,0 +1,119 @@
|
||||
package source
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"math"
|
||||
"reflect"
|
||||
"sort"
|
||||
)
|
||||
|
||||
// CloneMetadata returns an independently owned copy of JSON-shaped metadata.
|
||||
// It preserves concrete map, slice, and array types while rejecting values that
|
||||
// cannot be safely represented as JSON-shaped metadata.
|
||||
func CloneMetadata(metadata map[string]any) (map[string]any, error) {
|
||||
if len(metadata) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
cloned, err := cloneMetadataValue(reflect.ValueOf(metadata), "metadata", make(map[metadataVisit]struct{}))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return cloned.Interface().(map[string]any), nil
|
||||
}
|
||||
|
||||
type metadataVisit struct {
|
||||
typ reflect.Type
|
||||
ptr uintptr
|
||||
}
|
||||
|
||||
func cloneMetadataValue(value reflect.Value, location string, active map[metadataVisit]struct{}) (reflect.Value, error) {
|
||||
if !value.IsValid() {
|
||||
return value, nil
|
||||
}
|
||||
|
||||
switch value.Kind() {
|
||||
case reflect.Interface:
|
||||
if value.IsNil() {
|
||||
return reflect.Zero(value.Type()), nil
|
||||
}
|
||||
cloned, err := cloneMetadataValue(value.Elem(), location, active)
|
||||
if err != nil {
|
||||
return reflect.Value{}, err
|
||||
}
|
||||
result := reflect.New(value.Type()).Elem()
|
||||
result.Set(cloned)
|
||||
return result, nil
|
||||
case reflect.Map:
|
||||
if value.IsNil() {
|
||||
return reflect.Zero(value.Type()), nil
|
||||
}
|
||||
if value.Type().Key().Kind() != reflect.String {
|
||||
return reflect.Value{}, fmt.Errorf("%s has unsupported map key type %s", location, value.Type().Key())
|
||||
}
|
||||
leave, err := enterMetadataValue(value, active, location)
|
||||
if err != nil {
|
||||
return reflect.Value{}, err
|
||||
}
|
||||
defer leave()
|
||||
|
||||
keys := value.MapKeys()
|
||||
sort.Slice(keys, func(i, j int) bool { return keys[i].String() < keys[j].String() })
|
||||
result := reflect.MakeMapWithSize(value.Type(), value.Len())
|
||||
for _, key := range keys {
|
||||
cloned, err := cloneMetadataValue(value.MapIndex(key), location+"."+key.String(), active)
|
||||
if err != nil {
|
||||
return reflect.Value{}, err
|
||||
}
|
||||
result.SetMapIndex(key, cloned)
|
||||
}
|
||||
return result, nil
|
||||
case reflect.Slice:
|
||||
if value.IsNil() {
|
||||
return reflect.Zero(value.Type()), nil
|
||||
}
|
||||
leave, err := enterMetadataValue(value, active, location)
|
||||
if err != nil {
|
||||
return reflect.Value{}, err
|
||||
}
|
||||
defer leave()
|
||||
|
||||
result := reflect.MakeSlice(value.Type(), value.Len(), value.Len())
|
||||
for i := 0; i < value.Len(); i++ {
|
||||
cloned, err := cloneMetadataValue(value.Index(i), fmt.Sprintf("%s[%d]", location, i), active)
|
||||
if err != nil {
|
||||
return reflect.Value{}, err
|
||||
}
|
||||
result.Index(i).Set(cloned)
|
||||
}
|
||||
return result, nil
|
||||
case reflect.Array:
|
||||
result := reflect.New(value.Type()).Elem()
|
||||
for i := 0; i < value.Len(); i++ {
|
||||
cloned, err := cloneMetadataValue(value.Index(i), fmt.Sprintf("%s[%d]", location, i), active)
|
||||
if err != nil {
|
||||
return reflect.Value{}, err
|
||||
}
|
||||
result.Index(i).Set(cloned)
|
||||
}
|
||||
return result, nil
|
||||
case reflect.Float32, reflect.Float64:
|
||||
if math.IsNaN(value.Float()) || math.IsInf(value.Float(), 0) {
|
||||
return reflect.Value{}, fmt.Errorf("%s has a non-finite number", location)
|
||||
}
|
||||
return value, nil
|
||||
case reflect.Bool, reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64,
|
||||
reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64, reflect.String:
|
||||
return value, nil
|
||||
default:
|
||||
return reflect.Value{}, fmt.Errorf("%s has unsupported type %s", location, value.Type())
|
||||
}
|
||||
}
|
||||
|
||||
func enterMetadataValue(value reflect.Value, active map[metadataVisit]struct{}, location string) (func(), error) {
|
||||
visit := metadataVisit{typ: value.Type(), ptr: value.Pointer()}
|
||||
if _, exists := active[visit]; exists {
|
||||
return nil, fmt.Errorf("%s contains a cycle", location)
|
||||
}
|
||||
active[visit] = struct{}{}
|
||||
return func() { delete(active, visit) }, nil
|
||||
}
|
||||
@@ -2,8 +2,10 @@ package chunkplan
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
@@ -16,9 +18,11 @@ import (
|
||||
|
||||
const SchemaVersion = pipeline.ChunkPlanSchemaVersion
|
||||
|
||||
const planFileName = "plan.json"
|
||||
|
||||
type filesystemStore struct {
|
||||
root string
|
||||
write func(string, []byte) error
|
||||
write func(*os.Root, string, []byte) error
|
||||
}
|
||||
|
||||
func NewFilesystemStore(root string) (pipeline.ChunkPlanStore, error) {
|
||||
@@ -33,14 +37,46 @@ func NewFilesystemStore(root string) (pipeline.ChunkPlanStore, error) {
|
||||
}
|
||||
|
||||
func (s *filesystemStore) Load(sourceDigest string) (pipeline.ChunkPlanRecord, pipeline.ChunkPlanDecision, error) {
|
||||
target, err := s.planPath(sourceDigest)
|
||||
digestDir, err := digestPathSegment(sourceDigest)
|
||||
if err != nil {
|
||||
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{}, err
|
||||
}
|
||||
data, err := os.ReadFile(target)
|
||||
root, err := s.openRoot(false)
|
||||
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{Status: pipeline.ChunkPlanMissing, Reason: lookupReason(pipeline.ChunkPlanMissing)}, nil
|
||||
}
|
||||
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{}, fmt.Errorf("open chunk plan root: %w", err)
|
||||
}
|
||||
defer root.Close()
|
||||
|
||||
state, err := inspectDirectory(root, digestDir)
|
||||
if err != nil {
|
||||
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{}, fmt.Errorf("inspect chunk plan directory: %w", err)
|
||||
}
|
||||
if state == entryMissing {
|
||||
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{Status: pipeline.ChunkPlanMissing, Reason: lookupReason(pipeline.ChunkPlanMissing)}, nil
|
||||
}
|
||||
if state == entryRejected {
|
||||
return invalidDecision()
|
||||
}
|
||||
|
||||
target := planPath(digestDir)
|
||||
state, err = inspectPlan(root, target)
|
||||
if err != nil {
|
||||
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{}, fmt.Errorf("inspect chunk plan file: %w", err)
|
||||
}
|
||||
if state == entryMissing {
|
||||
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{Status: pipeline.ChunkPlanMissing, Reason: lookupReason(pipeline.ChunkPlanMissing)}, nil
|
||||
}
|
||||
if state == entryRejected {
|
||||
return invalidDecision()
|
||||
}
|
||||
|
||||
data, err := root.ReadFile(target)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{Status: pipeline.ChunkPlanMissing, Reason: lookupReason(pipeline.ChunkPlanMissing)}, nil
|
||||
}
|
||||
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{}, fmt.Errorf("read chunk plan: %w", err)
|
||||
}
|
||||
@@ -50,26 +86,23 @@ func (s *filesystemStore) Load(sourceDigest string) (pipeline.ChunkPlanRecord, p
|
||||
decoder.DisallowUnknownFields()
|
||||
decoder.UseNumber()
|
||||
if err := decoder.Decode(&record); err != nil {
|
||||
return invalidDecision(fmt.Sprintf("decode stored chunk plan: %v", err))
|
||||
return invalidDecision()
|
||||
}
|
||||
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))
|
||||
return invalidDecision()
|
||||
}
|
||||
if err := validateRecord(record, sourceDigest); err != nil {
|
||||
return invalidDecision(err.Error())
|
||||
return invalidDecision()
|
||||
}
|
||||
return record, pipeline.ChunkPlanDecision{Status: pipeline.ChunkPlanHit, Reason: "stored chunk plan is valid"}, nil
|
||||
return record, pipeline.ChunkPlanDecision{Status: pipeline.ChunkPlanHit, Reason: lookupReason(pipeline.ChunkPlanHit)}, 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)
|
||||
digestDir, err := digestPathSegment(record.SourceDigest)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -78,25 +111,134 @@ func (s *filesystemStore) Save(record pipeline.ChunkPlanRecord) error {
|
||||
return fmt.Errorf("encode chunk plan record: %w", err)
|
||||
}
|
||||
data = append(data, '\n')
|
||||
root, err := s.openRoot(true)
|
||||
if err != nil {
|
||||
return fmt.Errorf("open chunk plan root: %w", err)
|
||||
}
|
||||
defer root.Close()
|
||||
if err := ensureDirectory(root, digestDir); err != nil {
|
||||
return fmt.Errorf("prepare chunk plan directory: %w", err)
|
||||
}
|
||||
target := planPath(digestDir)
|
||||
state, err := inspectPlan(root, target)
|
||||
if err != nil {
|
||||
return fmt.Errorf("inspect chunk plan file: %w", err)
|
||||
}
|
||||
if state == entryRejected {
|
||||
return fmt.Errorf("chunk plan file has an unsupported type")
|
||||
}
|
||||
writer := s.write
|
||||
if writer == nil {
|
||||
writer = writeAtomic
|
||||
}
|
||||
if err := writer(target, data); err != nil {
|
||||
if err := writer(root, target, data); err != nil {
|
||||
return fmt.Errorf("write chunk plan: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *filesystemStore) planPath(sourceDigest string) (string, error) {
|
||||
func (s *filesystemStore) openRoot(create bool) (*os.Root, error) {
|
||||
if s == nil || strings.TrimSpace(s.root) == "" {
|
||||
return "", fmt.Errorf("chunk plan store must not be nil")
|
||||
return nil, fmt.Errorf("chunk plan store must not be nil")
|
||||
}
|
||||
hexDigest, err := digestPathSegment(sourceDigest)
|
||||
if create {
|
||||
if err := os.MkdirAll(s.root, 0o700); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
root, err := os.OpenRoot(s.root)
|
||||
if err != nil {
|
||||
return "", err
|
||||
return nil, err
|
||||
}
|
||||
return filepath.Join(s.root, hexDigest, "plan.json"), nil
|
||||
if !create {
|
||||
return root, nil
|
||||
}
|
||||
rootDirectory, err := root.Open(".")
|
||||
if err != nil {
|
||||
_ = root.Close()
|
||||
return nil, err
|
||||
}
|
||||
defer rootDirectory.Close()
|
||||
if err := rootDirectory.Chmod(0o700); err != nil {
|
||||
_ = root.Close()
|
||||
return nil, err
|
||||
}
|
||||
return root, nil
|
||||
}
|
||||
|
||||
type entryState uint8
|
||||
|
||||
const (
|
||||
entryPresent entryState = iota
|
||||
entryMissing
|
||||
entryRejected
|
||||
)
|
||||
|
||||
func inspectDirectory(root *os.Root, digestDir string) (entryState, error) {
|
||||
info, err := root.Lstat(digestDir)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
return entryMissing, nil
|
||||
}
|
||||
return entryPresent, err
|
||||
}
|
||||
if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() {
|
||||
return entryRejected, nil
|
||||
}
|
||||
return entryPresent, nil
|
||||
}
|
||||
|
||||
func ensureDirectory(root *os.Root, digestDir string) error {
|
||||
for {
|
||||
state, err := inspectDirectory(root, digestDir)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
switch state {
|
||||
case entryRejected:
|
||||
return fmt.Errorf("chunk plan directory has an unsupported type")
|
||||
case entryMissing:
|
||||
if err := root.Mkdir(digestDir, 0o700); err != nil && !errors.Is(err, os.ErrExist) {
|
||||
return err
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
directory, err := root.Open(digestDir)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
info, statErr := directory.Stat()
|
||||
if statErr == nil && !info.IsDir() {
|
||||
statErr = fmt.Errorf("chunk plan directory has an unsupported type")
|
||||
}
|
||||
if statErr == nil {
|
||||
statErr = directory.Chmod(0o700)
|
||||
}
|
||||
closeErr := directory.Close()
|
||||
if statErr != nil {
|
||||
return statErr
|
||||
}
|
||||
return closeErr
|
||||
}
|
||||
}
|
||||
|
||||
func inspectPlan(root *os.Root, target string) (entryState, error) {
|
||||
info, err := root.Lstat(target)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
return entryMissing, nil
|
||||
}
|
||||
return entryPresent, err
|
||||
}
|
||||
if info.Mode()&os.ModeSymlink != 0 || !info.Mode().IsRegular() {
|
||||
return entryRejected, nil
|
||||
}
|
||||
return entryPresent, nil
|
||||
}
|
||||
|
||||
func planPath(digestDir string) string {
|
||||
return digestDir + "/" + planFileName
|
||||
}
|
||||
|
||||
func digestPathSegment(digest string) (string, error) {
|
||||
@@ -165,8 +307,21 @@ func validateRecord(record pipeline.ChunkPlanRecord, requestedDigest string) err
|
||||
return nil
|
||||
}
|
||||
|
||||
func invalidDecision(reason string) (pipeline.ChunkPlanRecord, pipeline.ChunkPlanDecision, error) {
|
||||
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{Status: pipeline.ChunkPlanInvalid, Reason: reason}, nil
|
||||
func invalidDecision() (pipeline.ChunkPlanRecord, pipeline.ChunkPlanDecision, error) {
|
||||
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{Status: pipeline.ChunkPlanInvalid, Reason: lookupReason(pipeline.ChunkPlanInvalid)}, nil
|
||||
}
|
||||
|
||||
func lookupReason(status pipeline.ChunkPlanStatus) string {
|
||||
switch status {
|
||||
case pipeline.ChunkPlanHit:
|
||||
return "stored chunk plan is valid"
|
||||
case pipeline.ChunkPlanMissing:
|
||||
return "chunk plan not found"
|
||||
case pipeline.ChunkPlanInvalid:
|
||||
return "stored chunk plan is invalid"
|
||||
default:
|
||||
return "chunk plan lookup skipped"
|
||||
}
|
||||
}
|
||||
|
||||
type atomicWriteHooks struct {
|
||||
@@ -174,33 +329,24 @@ type atomicWriteHooks struct {
|
||||
BeforeRename func() error
|
||||
}
|
||||
|
||||
func writeAtomic(target string, data []byte) error {
|
||||
return writeAtomicWithHooks(target, data, atomicWriteHooks{})
|
||||
func writeAtomic(root *os.Root, target string, data []byte) error {
|
||||
return writeAtomicWithHooks(root, target, data, atomicWriteHooks{})
|
||||
}
|
||||
|
||||
func writeAtomicWithHooks(target string, data []byte, hooks atomicWriteHooks) 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
|
||||
}
|
||||
|
||||
func writeAtomicWithHooks(root *os.Root, target string, data []byte, hooks atomicWriteHooks) error {
|
||||
if hooks.BeforeCreateTemp != nil {
|
||||
if err := hooks.BeforeCreateTemp(); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
temp, err := os.CreateTemp(dir, ".plan.json.tmp-*")
|
||||
temp, tempPath, err := createTemporaryFile(root, target)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
tempPath := temp.Name()
|
||||
removeTemp := true
|
||||
defer func() {
|
||||
if removeTemp {
|
||||
_ = os.Remove(tempPath)
|
||||
_ = root.Remove(tempPath)
|
||||
}
|
||||
}()
|
||||
if err := temp.Chmod(0o600); err != nil {
|
||||
@@ -223,9 +369,28 @@ func writeAtomicWithHooks(target string, data []byte, hooks atomicWriteHooks) er
|
||||
return err
|
||||
}
|
||||
}
|
||||
if err := os.Rename(tempPath, target); err != nil {
|
||||
if err := root.Rename(tempPath, target); err != nil {
|
||||
return err
|
||||
}
|
||||
removeTemp = false
|
||||
return nil
|
||||
}
|
||||
|
||||
func createTemporaryFile(root *os.Root, target string) (*os.File, string, error) {
|
||||
for attempt := 0; attempt < 32; attempt++ {
|
||||
var suffix [16]byte
|
||||
if _, err := rand.Read(suffix[:]); err != nil {
|
||||
return nil, "", err
|
||||
}
|
||||
path := strings.TrimSuffix(target, planFileName) + ".plan.json.tmp-" + hex.EncodeToString(suffix[:])
|
||||
file, err := root.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600)
|
||||
if errors.Is(err, os.ErrExist) {
|
||||
continue
|
||||
}
|
||||
if err != nil {
|
||||
return nil, "", err
|
||||
}
|
||||
return file, path, nil
|
||||
}
|
||||
return nil, "", fmt.Errorf("create unique temporary chunk plan file")
|
||||
}
|
||||
|
||||
@@ -84,19 +84,118 @@ func TestFilesystemStorePermissions(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestFilesystemStoreMissingAndOperationalErrors(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
root := filepath.Join(t.TempDir(), "plans")
|
||||
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)
|
||||
}
|
||||
if _, err := os.Stat(root); !os.IsNotExist(err) {
|
||||
t.Fatalf("Load() created missing root: %v", 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")
|
||||
_, decision, err = store.Load(testSourceDigest)
|
||||
if err != nil || decision.Status != pipeline.ChunkPlanInvalid {
|
||||
t.Fatalf("Load(plan.json directory) decision=%#v error=%v", decision, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestFilesystemStoreRejectsSymlinkedEntries(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
setup func(t *testing.T, root, outside string)
|
||||
}{
|
||||
{
|
||||
name: "digest directory",
|
||||
setup: func(t *testing.T, root, outside string) {
|
||||
t.Helper()
|
||||
if err := os.MkdirAll(outside, 0o700); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
writeFile(t, filepath.Join(outside, "plan.json"), []byte("outside plan"), 0o640)
|
||||
createSymlink(t, outside, filepath.Join(root, strings.TrimPrefix(testSourceDigest, "sha256:")))
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "plan file",
|
||||
setup: func(t *testing.T, root, outside string) {
|
||||
t.Helper()
|
||||
digestDir := filepath.Join(root, strings.TrimPrefix(testSourceDigest, "sha256:"))
|
||||
if err := os.MkdirAll(digestDir, 0o700); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
writeFile(t, outside, []byte("outside plan"), 0o640)
|
||||
createSymlink(t, outside, filepath.Join(digestDir, "plan.json"))
|
||||
},
|
||||
},
|
||||
}
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
root := filepath.Join(t.TempDir(), "plans")
|
||||
if err := os.MkdirAll(root, 0o700); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
outside := filepath.Join(t.TempDir(), "outside")
|
||||
tc.setup(t, root, outside)
|
||||
before := readFile(t, outsidePlanPath(tc.name, outside))
|
||||
beforeMode := fileMode(t, outsidePlanPath(tc.name, outside))
|
||||
|
||||
store := newStore(t, root)
|
||||
_, decision, err := store.Load(testSourceDigest)
|
||||
if err != nil || decision.Status != pipeline.ChunkPlanInvalid {
|
||||
t.Fatalf("Load() decision=%#v error=%v", decision, err)
|
||||
}
|
||||
if err := store.Save(testRecord(t, 1)); err == nil {
|
||||
t.Fatal("Save() error = nil")
|
||||
}
|
||||
outsidePlan := outsidePlanPath(tc.name, outside)
|
||||
if got := readFile(t, outsidePlan); !bytes.Equal(got, before) {
|
||||
t.Fatalf("outside content = %q, want %q", got, before)
|
||||
}
|
||||
if got := fileMode(t, outsidePlan); got != beforeMode {
|
||||
t.Fatalf("outside mode = %04o, want %04o", got, beforeMode)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestFilesystemStoreRejectsUnexpectedEntryTypes(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
setup func(t *testing.T, root string)
|
||||
}{
|
||||
{
|
||||
name: "digest file",
|
||||
setup: func(t *testing.T, root string) {
|
||||
writeFile(t, filepath.Join(root, strings.TrimPrefix(testSourceDigest, "sha256:")), []byte("not a directory"), 0o600)
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "plan directory",
|
||||
setup: func(t *testing.T, root string) {
|
||||
if err := os.MkdirAll(filepath.Join(root, strings.TrimPrefix(testSourceDigest, "sha256:"), "plan.json"), 0o700); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
},
|
||||
},
|
||||
}
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
tc.setup(t, root)
|
||||
store := newStore(t, root)
|
||||
_, decision, err := store.Load(testSourceDigest)
|
||||
if err != nil || decision.Status != pipeline.ChunkPlanInvalid {
|
||||
t.Fatalf("Load() decision=%#v error=%v", decision, err)
|
||||
}
|
||||
if err := store.Save(testRecord(t, 1)); err == nil {
|
||||
t.Fatal("Save() error = nil")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -122,14 +221,13 @@ 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: "truncated JSON", mutate: func(data []byte) []byte { return data[:len(data)/2] }, want: "decode"},
|
||||
{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"},
|
||||
return bytes.Replace(data, []byte(`{"schema_version"`), []byte(`{"SENTINEL_UNKNOWN_FIELD":true,"schema_version"`), 1)
|
||||
}},
|
||||
{name: "truncated JSON", mutate: func(data []byte) []byte { return data[:len(data)/2] }},
|
||||
{name: "schema mismatch", mutate: replaceJSON(`notarius.chunk-plan.v1`, `SENTINEL_SCHEMA_VALUE`)},
|
||||
{name: "source mismatch", mutate: replaceJSON(testSourceDigest, "sha256:"+strings.Repeat("b", 64))},
|
||||
{name: "plan digest mismatch", mutate: func(data []byte) []byte {
|
||||
prefix := []byte(`"plan_digest":"sha256:`)
|
||||
index := bytes.Index(data, prefix)
|
||||
@@ -137,14 +235,15 @@ func TestFilesystemStoreReportsInvalidRecordsAsRecoverable(t *testing.T) {
|
||||
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"},
|
||||
return bytes.Replace(data, []byte(`"test/value"`), []byte(`"SENTINEL_ANNOTATION_NAMESPACE"`), 1)
|
||||
}},
|
||||
{name: "bad boundary", mutate: func(data []byte) []byte {
|
||||
return bytes.Replace(data, []byte(`"start_unit_id":1`), []byte(`"start_unit_id":0`), 1)
|
||||
}, want: "boundaries must be positive"},
|
||||
{name: "trailing JSON", mutate: func(data []byte) []byte { return append(data, []byte(` {}`)...) }, want: "trailing"},
|
||||
}},
|
||||
{name: "timestamp", mutate: replaceJSON(`2026-07-18T12:00:00Z`, `SENTINEL_TIMESTAMP`)},
|
||||
{name: "trailing JSON", mutate: func(data []byte) []byte { return append(data, []byte(` {"SENTINEL_TRAILING":true}`)...) }},
|
||||
}
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
@@ -163,9 +262,14 @@ func TestFilesystemStoreReportsInvalidRecordsAsRecoverable(t *testing.T) {
|
||||
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) {
|
||||
if err != nil || decision.Status != pipeline.ChunkPlanInvalid || !reflect.DeepEqual(got, pipeline.ChunkPlanRecord{}) || decision.Reason != "stored chunk plan is invalid" {
|
||||
t.Fatalf("record=%#v decision=%#v error=%v", got, decision, err)
|
||||
}
|
||||
for _, sentinel := range []string{"SENTINEL_UNKNOWN_FIELD", "SENTINEL_SCHEMA_VALUE", "SENTINEL_ANNOTATION_NAMESPACE", "SENTINEL_TIMESTAMP", "SENTINEL_TRAILING"} {
|
||||
if strings.Contains(decision.Reason, sentinel) {
|
||||
t.Fatalf("decision leaked %q: %#v", sentinel, decision)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -290,7 +394,9 @@ func TestFilesystemStoreInterruptedWritesPreservePreviousRecord(t *testing.T) {
|
||||
{name: "before rename", hooks: atomicWriteHooks{BeforeRename: func() error { return errors.New("interrupted before rename") }}},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
store.write = func(target string, data []byte) error { return writeAtomicWithHooks(target, data, tc.hooks) }
|
||||
store.write = func(root *os.Root, target string, data []byte) error {
|
||||
return writeAtomicWithHooks(root, target, data, tc.hooks)
|
||||
}
|
||||
if err := store.Save(testRecord(t, 2)); err == nil {
|
||||
t.Fatal("Save() error = nil")
|
||||
}
|
||||
@@ -358,3 +464,45 @@ func assertNoTemps(t *testing.T, dir string) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func createSymlink(t *testing.T, target, link string) {
|
||||
t.Helper()
|
||||
if err := os.Symlink(target, link); err != nil {
|
||||
t.Skipf("create symlink: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func writeFile(t *testing.T, path string, data []byte, mode os.FileMode) {
|
||||
t.Helper()
|
||||
if err := os.WriteFile(path, data, mode); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.Chmod(path, mode); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func readFile(t *testing.T, path string) []byte {
|
||||
t.Helper()
|
||||
data, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return data
|
||||
}
|
||||
|
||||
func fileMode(t *testing.T, path string) os.FileMode {
|
||||
t.Helper()
|
||||
info, err := os.Stat(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return info.Mode().Perm()
|
||||
}
|
||||
|
||||
func outsidePlanPath(name, outside string) string {
|
||||
if name == "digest directory" {
|
||||
return filepath.Join(outside, "plan.json")
|
||||
}
|
||||
return outside
|
||||
}
|
||||
|
||||
@@ -679,41 +679,11 @@ func validateOutputFileName(name string) error {
|
||||
}
|
||||
|
||||
func cloneMetadata(metadata map[string]any) map[string]any {
|
||||
if len(metadata) == 0 {
|
||||
return nil
|
||||
}
|
||||
out := make(map[string]any, len(metadata))
|
||||
for key, value := range metadata {
|
||||
out[key] = cloneMetadataValue(value)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func cloneMetadataValue(value any) any {
|
||||
switch typed := value.(type) {
|
||||
case map[string]any:
|
||||
return cloneMetadata(typed)
|
||||
case []any:
|
||||
out := make([]any, len(typed))
|
||||
for i := range typed {
|
||||
out[i] = cloneMetadataValue(typed[i])
|
||||
}
|
||||
return out
|
||||
case json.RawMessage:
|
||||
return append(json.RawMessage(nil), typed...)
|
||||
case []byte:
|
||||
return append([]byte(nil), typed...)
|
||||
case []string:
|
||||
return append([]string(nil), typed...)
|
||||
case map[string]string:
|
||||
out := make(map[string]string, len(typed))
|
||||
for key, item := range typed {
|
||||
out[key] = item
|
||||
}
|
||||
return out
|
||||
default:
|
||||
return value
|
||||
cloned, err := source.CloneMetadata(metadata)
|
||||
if err != nil {
|
||||
return metadata
|
||||
}
|
||||
return cloned
|
||||
}
|
||||
|
||||
func cloneLLMProfiles(profiles []artifacts.LLMProfileManifest) []artifacts.LLMProfileManifest {
|
||||
|
||||
@@ -47,11 +47,11 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S
|
||||
if mode == ChunkCacheAuto {
|
||||
record, decision, err := input.ChunkPlans.Load(doc.Digest)
|
||||
result.lookup = decision
|
||||
result.summary.LookupStatus = string(decision.Status)
|
||||
result.summary.LookupReason = decision.Reason
|
||||
result.summary.LookupStatus = chunkPlanLookupStatus(decision.Status)
|
||||
result.summary.LookupReason = chunkPlanLookupReason(decision.Status)
|
||||
if err != nil {
|
||||
result.summary.LookupStatus = "skipped"
|
||||
result.summary.LookupReason = "chunk plan lookup failed"
|
||||
result.summary.LookupReason = chunkPlanLookupReason("")
|
||||
return result, fmt.Errorf("load chunk plan: %w", err)
|
||||
}
|
||||
switch decision.Status {
|
||||
@@ -68,13 +68,13 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S
|
||||
result.setValidation(validationWarnings, rejection, err)
|
||||
return result, err
|
||||
}
|
||||
result.lookup = ChunkPlanDecision{Status: ChunkPlanInvalid, Reason: "stored chunk plan cannot be materialized against the current source"}
|
||||
result.lookup = ChunkPlanDecision{Status: ChunkPlanInvalid, Reason: chunkPlanLookupReason(ChunkPlanInvalid)}
|
||||
result.summary.LookupStatus = "invalid"
|
||||
result.summary.LookupReason = result.lookup.Reason
|
||||
result.summary.LookupReason = chunkPlanLookupReason(ChunkPlanInvalid)
|
||||
case ChunkPlanMissing, ChunkPlanInvalid:
|
||||
// Generate below.
|
||||
default:
|
||||
return result, fmt.Errorf("load chunk plan returned unsupported status %q", decision.Status)
|
||||
return result, fmt.Errorf("load chunk plan returned unsupported decision status")
|
||||
}
|
||||
}
|
||||
if mode == ChunkCacheAuto || mode == ChunkCacheRefresh {
|
||||
@@ -171,6 +171,32 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func chunkPlanLookupStatus(status ChunkPlanStatus) string {
|
||||
switch status {
|
||||
case ChunkPlanHit:
|
||||
return "hit"
|
||||
case ChunkPlanMissing:
|
||||
return "missing"
|
||||
case ChunkPlanInvalid:
|
||||
return "invalid"
|
||||
default:
|
||||
return "skipped"
|
||||
}
|
||||
}
|
||||
|
||||
func chunkPlanLookupReason(status ChunkPlanStatus) string {
|
||||
switch status {
|
||||
case ChunkPlanHit:
|
||||
return "stored chunk plan is valid"
|
||||
case ChunkPlanMissing:
|
||||
return "chunk plan not found"
|
||||
case ChunkPlanInvalid:
|
||||
return "stored chunk plan is invalid"
|
||||
default:
|
||||
return "chunk plan lookup skipped"
|
||||
}
|
||||
}
|
||||
|
||||
func (result *chunkPlanExecution) setCandidate(record ChunkPlanRecord, action string) {
|
||||
cloned := cloneChunkPlanRecord(record)
|
||||
result.record = &cloned
|
||||
|
||||
@@ -143,6 +143,43 @@ func TestRunnerChunkPlanModeMatrix(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunnerUsesStatusDerivedChunkPlanSummaryReasons(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
status ChunkPlanStatus
|
||||
wantStatus string
|
||||
wantReason string
|
||||
}{
|
||||
{name: "hit", status: ChunkPlanHit, wantStatus: "hit", wantReason: "stored chunk plan is valid"},
|
||||
{name: "missing", status: ChunkPlanMissing, wantStatus: "missing", wantReason: "chunk plan not found"},
|
||||
{name: "invalid", status: ChunkPlanInvalid, wantStatus: "invalid", wantReason: "stored chunk plan is invalid"},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
prepared, plan := preparedTerminalDebugPipeline(t)
|
||||
store := &recordingChunkPlanStore{
|
||||
record: chunkPlanRecord(t, prepared, plan),
|
||||
decision: ChunkPlanDecision{Status: tc.status, Reason: "SENTINEL_STORE_REASON"},
|
||||
}
|
||||
output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ChunkCacheMode: ChunkCacheAuto, ChunkPlans: store})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if output.ChunkPlan.LookupStatus != tc.wantStatus || output.ChunkPlan.LookupReason != tc.wantReason || strings.Contains(output.ChunkPlan.LookupReason, "SENTINEL_STORE_REASON") {
|
||||
t.Fatalf("summary = %#v", output.ChunkPlan)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
prepared, _ := preparedTerminalDebugPipeline(t)
|
||||
output, err := New().Run(context.Background(), RunInput{
|
||||
Prepared: prepared, RawInput: []byte("input"), ChunkCacheMode: ChunkCacheAuto,
|
||||
ChunkPlans: &recordingChunkPlanStore{loadErr: errors.New("read failed")},
|
||||
})
|
||||
if err == nil || output.ChunkPlan == nil || output.ChunkPlan.LookupStatus != "skipped" || output.ChunkPlan.LookupReason != "chunk plan lookup skipped" {
|
||||
t.Fatalf("operational lookup output=%#v error=%v", output.ChunkPlan, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunnerChunkPlanHitUsesStoredProducerProvenance(t *testing.T) {
|
||||
prepared, plan := preparedTerminalDebugPipeline(t)
|
||||
prepared.resolved.Chunk.Module = "chunk/requested"
|
||||
|
||||
@@ -16,6 +16,24 @@ type extractCaptureRecorder struct {
|
||||
checkpoint ExtractCheckpoint
|
||||
}
|
||||
|
||||
type pipelineTypedMetadata map[string]any
|
||||
|
||||
type mutatingMetadataValidator struct {
|
||||
seen string
|
||||
}
|
||||
|
||||
func (*mutatingMetadataValidator) Name() string { return "test/mutating-metadata" }
|
||||
func (*mutatingMetadataValidator) ExecutionClass() contracts.ExecutionClass {
|
||||
return contracts.ExecutionClassDeterministic
|
||||
}
|
||||
func (v *mutatingMetadataValidator) Validate(_ context.Context, request contracts.ChunkValidationRequest) (contracts.ValidationResult, error) {
|
||||
nested := request.Chunks[0].Units[0].Metadata["typed"].(pipelineTypedMetadata)
|
||||
v.seen = string(nested["raw"].(json.RawMessage))
|
||||
nested["raw"].(json.RawMessage)[0] = '['
|
||||
nested["changed"] = true
|
||||
return contracts.ValidationResult{Approved: true}, nil
|
||||
}
|
||||
|
||||
func (r *extractCaptureRecorder) ExtractSucceeded(_ string, _ string, _ []CheckpointFingerprint, outputs []CheckpointArtifact, rejected []contracts.RejectedOutput, warnings []contracts.Warning) error {
|
||||
r.checkpoint = ExtractCheckpoint{
|
||||
Outputs: cloneCheckpointArtifacts(outputs),
|
||||
@@ -77,6 +95,46 @@ func TestRunnerPassesIndependentAnnotationScopesToExtractors(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunnerPassesIndependentConcreteMetadataToValidatorsAndExtractors(t *testing.T) {
|
||||
prepared := preparedAttemptDebugPipeline(t)
|
||||
input := prepared.input.(*typedTestInput)
|
||||
input.doc.Units[0].Metadata = map[string]any{
|
||||
"typed": pipelineTypedMetadata{"raw": json.RawMessage(`{"value":true}`)},
|
||||
}
|
||||
var err error
|
||||
input.doc.Digest, err = source.DigestDocument(input.doc)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
chunker := prepared.chunker.(*typedTestChunker)
|
||||
chunker.plan = typedTestPlan(input.doc)
|
||||
|
||||
validator := &mutatingMetadataValidator{}
|
||||
prepared.chunkValidators.validators = []preparedValidator{{
|
||||
resolved: ResolvedValidator{Binding: Binding(validator.Name()), Target: ValidatorTargetChunk},
|
||||
chunk: validator,
|
||||
}}
|
||||
var extractorSeen string
|
||||
installExtractOperation(prepared, 0, func(_ context.Context, request contracts.TypedExtractionRequest) (erasedTypedResult, error) {
|
||||
nested := request.Chunk.Units[0].Metadata["typed"].(pipelineTypedMetadata)
|
||||
extractorSeen = string(nested["raw"].(json.RawMessage))
|
||||
nested["raw"].(json.RawMessage)[0] = '['
|
||||
nested["changed"] = true
|
||||
return erasedTypedResult{Value: typedValueForLane(0, request.Chunk.Index)}, nil
|
||||
})
|
||||
|
||||
if _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input")}); err != nil {
|
||||
t.Fatalf("Run() error = %v", err)
|
||||
}
|
||||
if validator.seen != `{"value":true}` || extractorSeen != `{"value":true}` {
|
||||
t.Fatalf("validator/extractor metadata = %q / %q", validator.seen, extractorSeen)
|
||||
}
|
||||
original := input.doc.Units[0].Metadata["typed"].(pipelineTypedMetadata)
|
||||
if _, changed := original["changed"]; changed || string(original["raw"].(json.RawMessage)) != `{"value":true}` {
|
||||
t.Fatalf("source metadata changed through runner handoff: %#v", original)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunnerContinuesFromFreshAndReusedExtractResults(t *testing.T) {
|
||||
prepared := preparedAttemptDebugPipeline(t)
|
||||
extractCalls := 0
|
||||
|
||||
Reference in New Issue
Block a user