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
|
decision, and publication decision. Its closed decision values make failures
|
||||||
and recoverable invalid records inspectable without serializing plan ranges,
|
and recoverable invalid records inspectable without serializing plan ranges,
|
||||||
annotations, source content, reference content, prompts, model responses, or
|
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
|
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
|
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
|
unit ranges and may carry optional namespaced JSON annotations; it does not
|
||||||
contain materialized chunk content. The framework canonicalizes annotations,
|
contain materialized chunk content. The framework canonicalizes annotations,
|
||||||
validates ranges against the current source, and materializes chunk IDs,
|
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
|
namespaces remain optional data: generic framework code and downstream modules
|
||||||
must not require D&D scene annotations or import `dnd/scenes`.
|
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/artifacts` | Run-manifest and provenance models. |
|
||||||
| `internal/core/config` | Defaults, YAML parsing, environment overrides, validation, redaction, and effective pipeline resolution. |
|
| `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/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. |
|
| `internal/core/workspace` | Effective workspace settings, confined paths and writes, and checkpoint identity and manifest models. |
|
||||||
|
|
||||||
## Framework Packages
|
## 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
|
digest, at least one range, existing ordered boundaries, and increasing range
|
||||||
starts. Ranges may overlap or leave gaps; a chunker may impose stricter policy.
|
starts. Ranges may overlap or leave gaps; a chunker may impose stricter policy.
|
||||||
Materialization deterministically reconstructs each range from the current
|
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'
|
Before lane execution, generic chunk validation checks the materialized chunks'
|
||||||
identities, order, source references, content, media type, units, and metadata.
|
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
|
succeeds. Delete an exact cache root or digest directory only when regeneration
|
||||||
cost is acceptable.
|
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
|
Publication uses atomic replacement. Concurrent readers observe a complete old
|
||||||
or new plan, and concurrent writers leave one complete valid winner; there is
|
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
|
no history, lock protocol, or rollback facility. Do not share a cache root
|
||||||
|
|||||||
@@ -1,9 +1,8 @@
|
|||||||
# ADR-0005 Feature Roadmap
|
# ADR-0005 Feature Roadmap
|
||||||
|
|
||||||
This roadmap defines the intended end state for
|
This roadmap records the implemented target state for
|
||||||
[ADR-0005](../adr/0005-cache-canonical-chunk-plans-by-source.md). The feature
|
[ADR-0005](../adr/0005-cache-canonical-chunk-plans-by-source.md). The
|
||||||
is not implemented. The ordered coding work belongs in the
|
[implementation record](implementation.md) preserves the completed work.
|
||||||
[implementation plan](implementation.md).
|
|
||||||
|
|
||||||
## Intent
|
## Intent
|
||||||
|
|
||||||
@@ -156,11 +155,11 @@ This feature does not provide:
|
|||||||
|
|
||||||
## Completion Outcomes
|
## 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,
|
reuse byte-stable materialized chunks across pipeline, lane, reference,
|
||||||
chunk-module, and LLM configuration changes; bypass and refresh obey their
|
chunk-module, and LLM configuration changes; bypass and refresh obey their
|
||||||
documented state semantics; provenance identifies the effective producer;
|
documented state semantics; provenance identifies the effective producer;
|
||||||
legacy chunk checkpoints cannot compete with plan reuse; and all focused,
|
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
|
configuration and operations references must also document both the per-user
|
||||||
default and the recommended system-wide Linux setting.
|
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 Feature Roadmap](adr0005.md), governed by
|
||||||
[ADR-0005](../adr/0005-cache-canonical-chunk-plans-by-source.md).
|
[ADR-0005](../adr/0005-cache-canonical-chunk-plans-by-source.md).
|
||||||
|
|
||||||
Stages 1 through 8 summarize the completed initial implementation. Stages 9
|
All stages are complete. This document remains a concise historical
|
||||||
through 12 track pending remediation identified by the post-implementation
|
implementation record. Current behavior is documented in the canonical
|
||||||
review. Current behavior is documented in the canonical references linked from
|
references linked from [Development](../development.md).
|
||||||
[Development](../development.md); those references must not describe the Stages
|
|
||||||
9 through 12 target state until the corresponding stage is complete.
|
|
||||||
|
|
||||||
## Execution Rules
|
## Execution Rules
|
||||||
|
|
||||||
@@ -351,7 +349,7 @@ feature and its per-user and system-wide deployment guidance.
|
|||||||
|
|
||||||
## Stage 9: Redact invalid cache-record diagnostics
|
## Stage 9: Redact invalid cache-record diagnostics
|
||||||
|
|
||||||
**Status:** Pending
|
**Status:** Complete
|
||||||
|
|
||||||
### Objective
|
### Objective
|
||||||
|
|
||||||
@@ -408,7 +406,7 @@ remain correctly classified, and all focused and repository-wide checks pass.
|
|||||||
|
|
||||||
## Stage 10: Enforce rooted chunk-plan filesystem access
|
## Stage 10: Enforce rooted chunk-plan filesystem access
|
||||||
|
|
||||||
**Status:** Pending
|
**Status:** Complete
|
||||||
|
|
||||||
### Objective
|
### 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
|
## Stage 11: Deep-clone source metadata during materialization
|
||||||
|
|
||||||
**Status:** Pending
|
**Status:** Complete
|
||||||
|
|
||||||
### Objective
|
### Objective
|
||||||
|
|
||||||
@@ -526,7 +524,7 @@ current materialized bytes remain stable; and all checks pass.
|
|||||||
|
|
||||||
## Stage 12: Finalize remediation documentation and roadmap status
|
## Stage 12: Finalize remediation documentation and roadmap status
|
||||||
|
|
||||||
**Status:** Pending
|
**Status:** Complete
|
||||||
|
|
||||||
### Objective
|
### 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) {
|
func TestDefaultAutoReusesPlanAcrossIndependentInvocations(t *testing.T) {
|
||||||
cacheBase := filepath.Join(t.TempDir(), "cache")
|
cacheBase := filepath.Join(t.TempDir(), "cache")
|
||||||
workspaceDir := filepath.Join(t.TempDir(), "workspace")
|
workspaceDir := filepath.Join(t.TempDir(), "workspace")
|
||||||
|
|||||||
@@ -162,7 +162,10 @@ func MaterializeChunkPlan(doc *SourceDocument, plan ChunkPlan) ([]Chunk, error)
|
|||||||
for index, chunkRange := range plan.Ranges {
|
for index, chunkRange := range plan.Ranges {
|
||||||
start, _ := UnitIndex(doc, chunkRange.StartUnitID)
|
start, _ := UnitIndex(doc, chunkRange.StartUnitID)
|
||||||
end, _ := UnitIndex(doc, chunkRange.EndUnitID)
|
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 {
|
content, err := json.Marshal(struct {
|
||||||
Units []SourceUnit `json:"units"`
|
Units []SourceUnit `json:"units"`
|
||||||
}{Units: units})
|
}{Units: units})
|
||||||
@@ -189,52 +192,18 @@ func MaterializeChunkPlan(doc *SourceDocument, plan ChunkPlan) ([]Chunk, error)
|
|||||||
return chunks, nil
|
return chunks, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func cloneSourceUnits(units []SourceUnit) []SourceUnit {
|
func cloneSourceUnits(units []SourceUnit) ([]SourceUnit, error) {
|
||||||
if len(units) == 0 {
|
if len(units) == 0 {
|
||||||
return nil
|
return nil, nil
|
||||||
}
|
}
|
||||||
cloned := make([]SourceUnit, len(units))
|
cloned := make([]SourceUnit, len(units))
|
||||||
for i, unit := range units {
|
for i, unit := range units {
|
||||||
cloned[i] = unit
|
cloned[i] = unit
|
||||||
cloned[i].Metadata = cloneJSONMap(unit.Metadata)
|
metadata, err := CloneMetadata(unit.Metadata)
|
||||||
}
|
if err != nil {
|
||||||
return cloned
|
return nil, fmt.Errorf("source unit[%d] metadata: %w", i, err)
|
||||||
}
|
|
||||||
|
|
||||||
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
|
cloned[i].Metadata = metadata
|
||||||
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
|
|
||||||
}
|
}
|
||||||
|
return cloned, nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -3,11 +3,16 @@ package source
|
|||||||
import (
|
import (
|
||||||
"bytes"
|
"bytes"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"math"
|
||||||
"reflect"
|
"reflect"
|
||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
type typedMetadataMap map[string]any
|
||||||
|
type typedMetadataSlice []typedMetadataMap
|
||||||
|
type typedMetadataArray [2]any
|
||||||
|
|
||||||
func TestCanonicalizeChunkAnnotations(t *testing.T) {
|
func TestCanonicalizeChunkAnnotations(t *testing.T) {
|
||||||
original := ChunkAnnotations{
|
original := ChunkAnnotations{
|
||||||
"domain/items": json.RawMessage(` { "z": [3, 2, 1], "a": 1.0 } `),
|
"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 {
|
func planDocument() *SourceDocument {
|
||||||
doc := &SourceDocument{ID: "source-plan", Kind: "test", Format: "application/test", Digest: "sha256:source-plan"}
|
doc := &SourceDocument{ID: "source-plan", Kind: "test", Format: "application/test", Digest: "sha256:source-plan"}
|
||||||
for _, id := range []int{10, 20, 30, 40, 50} {
|
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 (
|
import (
|
||||||
"bytes"
|
"bytes"
|
||||||
|
"crypto/rand"
|
||||||
"encoding/hex"
|
"encoding/hex"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"os"
|
"os"
|
||||||
@@ -16,9 +18,11 @@ import (
|
|||||||
|
|
||||||
const SchemaVersion = pipeline.ChunkPlanSchemaVersion
|
const SchemaVersion = pipeline.ChunkPlanSchemaVersion
|
||||||
|
|
||||||
|
const planFileName = "plan.json"
|
||||||
|
|
||||||
type filesystemStore struct {
|
type filesystemStore struct {
|
||||||
root string
|
root string
|
||||||
write func(string, []byte) error
|
write func(*os.Root, string, []byte) error
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewFilesystemStore(root string) (pipeline.ChunkPlanStore, 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) {
|
func (s *filesystemStore) Load(sourceDigest string) (pipeline.ChunkPlanRecord, pipeline.ChunkPlanDecision, error) {
|
||||||
target, err := s.planPath(sourceDigest)
|
digestDir, err := digestPathSegment(sourceDigest)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{}, err
|
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{}, err
|
||||||
}
|
}
|
||||||
data, err := os.ReadFile(target)
|
root, err := s.openRoot(false)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if os.IsNotExist(err) {
|
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)
|
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.DisallowUnknownFields()
|
||||||
decoder.UseNumber()
|
decoder.UseNumber()
|
||||||
if err := decoder.Decode(&record); err != nil {
|
if err := decoder.Decode(&record); err != nil {
|
||||||
return invalidDecision(fmt.Sprintf("decode stored chunk plan: %v", err))
|
return invalidDecision()
|
||||||
}
|
}
|
||||||
var trailing any
|
var trailing any
|
||||||
if err := decoder.Decode(&trailing); err != io.EOF {
|
if err := decoder.Decode(&trailing); err != io.EOF {
|
||||||
if err == nil {
|
return invalidDecision()
|
||||||
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 {
|
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 {
|
func (s *filesystemStore) Save(record pipeline.ChunkPlanRecord) error {
|
||||||
if err := validateRecord(record, record.SourceDigest); err != nil {
|
if err := validateRecord(record, record.SourceDigest); err != nil {
|
||||||
return fmt.Errorf("validate chunk plan record: %w", err)
|
return fmt.Errorf("validate chunk plan record: %w", err)
|
||||||
}
|
}
|
||||||
target, err := s.planPath(record.SourceDigest)
|
digestDir, err := digestPathSegment(record.SourceDigest)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -78,25 +111,134 @@ func (s *filesystemStore) Save(record pipeline.ChunkPlanRecord) error {
|
|||||||
return fmt.Errorf("encode chunk plan record: %w", err)
|
return fmt.Errorf("encode chunk plan record: %w", err)
|
||||||
}
|
}
|
||||||
data = append(data, '\n')
|
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
|
writer := s.write
|
||||||
if writer == nil {
|
if writer == nil {
|
||||||
writer = writeAtomic
|
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 fmt.Errorf("write chunk plan: %w", err)
|
||||||
}
|
}
|
||||||
return nil
|
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) == "" {
|
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 {
|
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) {
|
func digestPathSegment(digest string) (string, error) {
|
||||||
@@ -165,8 +307,21 @@ func validateRecord(record pipeline.ChunkPlanRecord, requestedDigest string) err
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func invalidDecision(reason string) (pipeline.ChunkPlanRecord, pipeline.ChunkPlanDecision, error) {
|
func invalidDecision() (pipeline.ChunkPlanRecord, pipeline.ChunkPlanDecision, error) {
|
||||||
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{Status: pipeline.ChunkPlanInvalid, Reason: reason}, nil
|
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 {
|
type atomicWriteHooks struct {
|
||||||
@@ -174,33 +329,24 @@ type atomicWriteHooks struct {
|
|||||||
BeforeRename func() error
|
BeforeRename func() error
|
||||||
}
|
}
|
||||||
|
|
||||||
func writeAtomic(target string, data []byte) error {
|
func writeAtomic(root *os.Root, target string, data []byte) error {
|
||||||
return writeAtomicWithHooks(target, data, atomicWriteHooks{})
|
return writeAtomicWithHooks(root, target, data, atomicWriteHooks{})
|
||||||
}
|
}
|
||||||
|
|
||||||
func writeAtomicWithHooks(target string, data []byte, hooks atomicWriteHooks) error {
|
func writeAtomicWithHooks(root *os.Root, 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
|
|
||||||
}
|
|
||||||
|
|
||||||
if hooks.BeforeCreateTemp != nil {
|
if hooks.BeforeCreateTemp != nil {
|
||||||
if err := hooks.BeforeCreateTemp(); err != nil {
|
if err := hooks.BeforeCreateTemp(); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
temp, err := os.CreateTemp(dir, ".plan.json.tmp-*")
|
temp, tempPath, err := createTemporaryFile(root, target)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
tempPath := temp.Name()
|
|
||||||
removeTemp := true
|
removeTemp := true
|
||||||
defer func() {
|
defer func() {
|
||||||
if removeTemp {
|
if removeTemp {
|
||||||
_ = os.Remove(tempPath)
|
_ = root.Remove(tempPath)
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
if err := temp.Chmod(0o600); err != nil {
|
if err := temp.Chmod(0o600); err != nil {
|
||||||
@@ -223,9 +369,28 @@ func writeAtomicWithHooks(target string, data []byte, hooks atomicWriteHooks) er
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if err := os.Rename(tempPath, target); err != nil {
|
if err := root.Rename(tempPath, target); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
removeTemp = false
|
removeTemp = false
|
||||||
return nil
|
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) {
|
func TestFilesystemStoreMissingAndOperationalErrors(t *testing.T) {
|
||||||
root := t.TempDir()
|
root := filepath.Join(t.TempDir(), "plans")
|
||||||
store := newStore(t, root)
|
store := newStore(t, root)
|
||||||
_, decision, err := store.Load(testSourceDigest)
|
_, decision, err := store.Load(testSourceDigest)
|
||||||
if err != nil || decision.Status != pipeline.ChunkPlanMissing {
|
if err != nil || decision.Status != pipeline.ChunkPlanMissing {
|
||||||
t.Fatalf("missing decision=%#v error=%v", decision, err)
|
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")
|
target := filepath.Join(root, strings.TrimPrefix(testSourceDigest, "sha256:"), "plan.json")
|
||||||
if err := os.MkdirAll(target, 0o700); err != nil {
|
if err := os.MkdirAll(target, 0o700); err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
if _, _, err := store.Load(testSourceDigest); err == nil {
|
_, decision, err = store.Load(testSourceDigest)
|
||||||
t.Fatal("Load(plan.json directory) error = nil")
|
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 {
|
tests := []struct {
|
||||||
name string
|
name string
|
||||||
mutate func([]byte) []byte
|
mutate func([]byte) []byte
|
||||||
want string
|
|
||||||
}{
|
}{
|
||||||
{name: "unknown field", mutate: func(data []byte) []byte {
|
{name: "unknown field", mutate: func(data []byte) []byte {
|
||||||
return bytes.Replace(data, []byte(`{"schema_version"`), []byte(`{"unknown":true,"schema_version"`), 1)
|
return bytes.Replace(data, []byte(`{"schema_version"`), []byte(`{"SENTINEL_UNKNOWN_FIELD":true,"schema_version"`), 1)
|
||||||
}, want: "unknown"},
|
}},
|
||||||
{name: "truncated JSON", mutate: func(data []byte) []byte { return data[:len(data)/2] }, want: "decode"},
|
{name: "truncated JSON", mutate: func(data []byte) []byte { return data[:len(data)/2] }},
|
||||||
{name: "schema mismatch", mutate: replaceJSON(`notarius.chunk-plan.v1`, `notarius.chunk-plan.v2`), want: "schema_version"},
|
{name: "schema mismatch", mutate: replaceJSON(`notarius.chunk-plan.v1`, `SENTINEL_SCHEMA_VALUE`)},
|
||||||
{name: "source mismatch", mutate: replaceJSON(testSourceDigest, "sha256:"+strings.Repeat("b", 64)), want: "source_digest"},
|
{name: "source mismatch", mutate: replaceJSON(testSourceDigest, "sha256:"+strings.Repeat("b", 64))},
|
||||||
{name: "plan digest mismatch", mutate: func(data []byte) []byte {
|
{name: "plan digest mismatch", mutate: func(data []byte) []byte {
|
||||||
prefix := []byte(`"plan_digest":"sha256:`)
|
prefix := []byte(`"plan_digest":"sha256:`)
|
||||||
index := bytes.Index(data, prefix)
|
index := bytes.Index(data, prefix)
|
||||||
@@ -137,14 +235,15 @@ func TestFilesystemStoreReportsInvalidRecordsAsRecoverable(t *testing.T) {
|
|||||||
data[index+len(prefix)] = '0'
|
data[index+len(prefix)] = '0'
|
||||||
}
|
}
|
||||||
return data
|
return data
|
||||||
}, want: "plan_digest"},
|
}},
|
||||||
{name: "noncanonical annotation", mutate: func(data []byte) []byte {
|
{name: "noncanonical annotation", mutate: func(data []byte) []byte {
|
||||||
return bytes.Replace(data, []byte(`{"value":1}`), []byte(`{ "value": 1 }`), 1)
|
return bytes.Replace(data, []byte(`"test/value"`), []byte(`"SENTINEL_ANNOTATION_NAMESPACE"`), 1)
|
||||||
}, want: "canonical JSON"},
|
}},
|
||||||
{name: "bad boundary", mutate: func(data []byte) []byte {
|
{name: "bad boundary", mutate: func(data []byte) []byte {
|
||||||
return bytes.Replace(data, []byte(`"start_unit_id":1`), []byte(`"start_unit_id":0`), 1)
|
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 {
|
for _, tc := range tests {
|
||||||
t.Run(tc.name, func(t *testing.T) {
|
t.Run(tc.name, func(t *testing.T) {
|
||||||
@@ -163,9 +262,14 @@ func TestFilesystemStoreReportsInvalidRecordsAsRecoverable(t *testing.T) {
|
|||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
got, decision, err := store.Load(testSourceDigest)
|
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)
|
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") }}},
|
{name: "before rename", hooks: atomicWriteHooks{BeforeRename: func() error { return errors.New("interrupted before rename") }}},
|
||||||
} {
|
} {
|
||||||
t.Run(tc.name, func(t *testing.T) {
|
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 {
|
if err := store.Save(testRecord(t, 2)); err == nil {
|
||||||
t.Fatal("Save() error = 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 {
|
func cloneMetadata(metadata map[string]any) map[string]any {
|
||||||
if len(metadata) == 0 {
|
cloned, err := source.CloneMetadata(metadata)
|
||||||
return nil
|
if err != nil {
|
||||||
}
|
return metadata
|
||||||
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
|
|
||||||
}
|
}
|
||||||
|
return cloned
|
||||||
}
|
}
|
||||||
|
|
||||||
func cloneLLMProfiles(profiles []artifacts.LLMProfileManifest) []artifacts.LLMProfileManifest {
|
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 {
|
if mode == ChunkCacheAuto {
|
||||||
record, decision, err := input.ChunkPlans.Load(doc.Digest)
|
record, decision, err := input.ChunkPlans.Load(doc.Digest)
|
||||||
result.lookup = decision
|
result.lookup = decision
|
||||||
result.summary.LookupStatus = string(decision.Status)
|
result.summary.LookupStatus = chunkPlanLookupStatus(decision.Status)
|
||||||
result.summary.LookupReason = decision.Reason
|
result.summary.LookupReason = chunkPlanLookupReason(decision.Status)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
result.summary.LookupStatus = "skipped"
|
result.summary.LookupStatus = "skipped"
|
||||||
result.summary.LookupReason = "chunk plan lookup failed"
|
result.summary.LookupReason = chunkPlanLookupReason("")
|
||||||
return result, fmt.Errorf("load chunk plan: %w", err)
|
return result, fmt.Errorf("load chunk plan: %w", err)
|
||||||
}
|
}
|
||||||
switch decision.Status {
|
switch decision.Status {
|
||||||
@@ -68,13 +68,13 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S
|
|||||||
result.setValidation(validationWarnings, rejection, err)
|
result.setValidation(validationWarnings, rejection, err)
|
||||||
return result, 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.LookupStatus = "invalid"
|
||||||
result.summary.LookupReason = result.lookup.Reason
|
result.summary.LookupReason = chunkPlanLookupReason(ChunkPlanInvalid)
|
||||||
case ChunkPlanMissing, ChunkPlanInvalid:
|
case ChunkPlanMissing, ChunkPlanInvalid:
|
||||||
// Generate below.
|
// Generate below.
|
||||||
default:
|
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 {
|
if mode == ChunkCacheAuto || mode == ChunkCacheRefresh {
|
||||||
@@ -171,6 +171,32 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S
|
|||||||
return result, nil
|
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) {
|
func (result *chunkPlanExecution) setCandidate(record ChunkPlanRecord, action string) {
|
||||||
cloned := cloneChunkPlanRecord(record)
|
cloned := cloneChunkPlanRecord(record)
|
||||||
result.record = &cloned
|
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) {
|
func TestRunnerChunkPlanHitUsesStoredProducerProvenance(t *testing.T) {
|
||||||
prepared, plan := preparedTerminalDebugPipeline(t)
|
prepared, plan := preparedTerminalDebugPipeline(t)
|
||||||
prepared.resolved.Chunk.Module = "chunk/requested"
|
prepared.resolved.Chunk.Module = "chunk/requested"
|
||||||
|
|||||||
@@ -16,6 +16,24 @@ type extractCaptureRecorder struct {
|
|||||||
checkpoint ExtractCheckpoint
|
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 {
|
func (r *extractCaptureRecorder) ExtractSucceeded(_ string, _ string, _ []CheckpointFingerprint, outputs []CheckpointArtifact, rejected []contracts.RejectedOutput, warnings []contracts.Warning) error {
|
||||||
r.checkpoint = ExtractCheckpoint{
|
r.checkpoint = ExtractCheckpoint{
|
||||||
Outputs: cloneCheckpointArtifacts(outputs),
|
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) {
|
func TestRunnerContinuesFromFreshAndReusedExtractResults(t *testing.T) {
|
||||||
prepared := preparedAttemptDebugPipeline(t)
|
prepared := preparedAttemptDebugPipeline(t)
|
||||||
extractCalls := 0
|
extractCalls := 0
|
||||||
|
|||||||
Reference in New Issue
Block a user