Preserve tier sequence operations in source history
This commit is contained in:
@@ -105,7 +105,14 @@ and declaring resource contexts.
|
|||||||
|
|
||||||
## 3. Apply Tier Sequence Operations To Historical Contributions — P2
|
## 3. Apply Tier Sequence Operations To Historical Contributions — P2
|
||||||
|
|
||||||
Status: [ ] Open.
|
Status: [x] Complete.
|
||||||
|
|
||||||
|
Completion: per-source history now composes strict item projections with the
|
||||||
|
tier operation, including nested application prefixes and local sequence replay.
|
||||||
|
Regressions cover all three operations, multiple peers, retained records, removed
|
||||||
|
fields, source attribution, and a subsequent overlay without a false decoding
|
||||||
|
event. Verified focused history tests, `go test -race ./internal/engine`, and
|
||||||
|
`./scripts/check.sh`.
|
||||||
|
|
||||||
Owner: [`applyOverlayWithCollectionPolicy`](../../internal/engine/overlay.go),
|
Owner: [`applyOverlayWithCollectionPolicy`](../../internal/engine/overlay.go),
|
||||||
lines 76–84, and existing sequence-history projection/replay.
|
lines 76–84, and existing sequence-history projection/replay.
|
||||||
|
|||||||
@@ -234,7 +234,7 @@ func applyHistoryApplication(root **document.Node, application historyApplicatio
|
|||||||
mergeHistoryDocumentAtPath(root, application.value, Path{}, index, application.sequencesByPath)
|
mergeHistoryDocumentAtPath(root, application.value, Path{}, index, application.sequencesByPath)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
mergeHistoryAtPath(root, application.path, application.value, index)
|
mergeHistoryAtPath(root, application.path, application.value, index, application.sequencesByPath)
|
||||||
}
|
}
|
||||||
|
|
||||||
func mergeHistoryDocument(target **document.Node, incoming *document.Node, index *mappingIndex) {
|
func mergeHistoryDocument(target **document.Node, incoming *document.Node, index *mappingIndex) {
|
||||||
@@ -351,9 +351,9 @@ func cloneSequenceHistoryContributions(values []sequenceHistoryContribution) []s
|
|||||||
return cloned
|
return cloned
|
||||||
}
|
}
|
||||||
|
|
||||||
func mergeHistoryAtPath(root **document.Node, path Path, incoming *document.Node, index *mappingIndex) {
|
func mergeHistoryAtPath(root **document.Node, path Path, incoming *document.Node, index *mappingIndex, sequences map[string]sequenceHistoryContributionIndex) {
|
||||||
if root == nil || *root == nil || len(path.segments) == 0 {
|
if root == nil || *root == nil || len(path.segments) == 0 {
|
||||||
mergeHistoryDocument(root, incoming, index)
|
mergeHistoryDocumentAtPath(root, incoming, Path{}, index, sequences)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
current := *root
|
current := *root
|
||||||
@@ -364,7 +364,7 @@ func mergeHistoryAtPath(root **document.Node, path Path, incoming *document.Node
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
if segmentIndex == len(path.segments)-1 {
|
if segmentIndex == len(path.segments)-1 {
|
||||||
mergeHistoryDocument(¤t.Sequence[position], incoming, index)
|
mergeHistoryDocumentAtPath(¤t.Sequence[position], incoming, Path{}, index, sequences)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
current = current.Sequence[position]
|
current = current.Sequence[position]
|
||||||
@@ -378,7 +378,7 @@ func mergeHistoryAtPath(root **document.Node, path Path, incoming *document.Node
|
|||||||
if position < 0 {
|
if position < 0 {
|
||||||
index.append(current, document.Entry{Key: segment.key, Value: document.Clone(incoming)})
|
index.append(current, document.Entry{Key: segment.key, Value: document.Clone(incoming)})
|
||||||
} else {
|
} else {
|
||||||
mergeHistoryDocument(¤t.Mapping[position].Value, incoming, index)
|
mergeHistoryDocumentAtPath(¤t.Mapping[position].Value, incoming, Path{}, index, sequences)
|
||||||
}
|
}
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -78,6 +78,71 @@ func TestSequenceHistoryReplaysAppendAtAssignedIndexes(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestTierSequenceHistoryReplaysSelectedOperation(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
policy := mustCompileSequenceIdentityPolicy(t, []sequenceIdentityRegistration{{
|
||||||
|
selector: sequenceIdentitySelector{literalSequenceSelector("records")}, identityFields: []string{"id"},
|
||||||
|
}})
|
||||||
|
for _, test := range []struct {
|
||||||
|
name string
|
||||||
|
mode sequenceMode
|
||||||
|
}{{"append", sequenceModeAppend}, {"upsert", sequenceModeUpsert}, {"replace", sequenceModeReplace}} {
|
||||||
|
t.Run(test.name, func(t *testing.T) {
|
||||||
|
mode := test.mode
|
||||||
|
lower, diagnostics := composeDocumentsWithPolicy(policy, parseCompositionDocument(t, "lower", "records: [{id: a, removed: old}, {id: b, value: old, removed: old}]\n"))
|
||||||
|
if len(diagnostics) != 0 {
|
||||||
|
t.Fatal(diagnostics)
|
||||||
|
}
|
||||||
|
secondID := "d"
|
||||||
|
if mode == sequenceModeUpsert {
|
||||||
|
secondID = "b"
|
||||||
|
}
|
||||||
|
declarations := []sequenceDeclarationSpec{{destination: Keys("records"), modePresent: true, mode: mode}}
|
||||||
|
higher, diagnostics := composeDocumentsWithCollectionPolicy(policy, lower.collectionPolicies, declarations,
|
||||||
|
parseCompositionDocument(t, "first-peer", "records: [{id: c, value: first}]\n"),
|
||||||
|
parseCompositionDocument(t, "second-peer", "records: [{id: "+secondID+", value: second}]\n"),
|
||||||
|
)
|
||||||
|
if len(diagnostics) != 0 {
|
||||||
|
t.Fatal(diagnostics)
|
||||||
|
}
|
||||||
|
assembly := newAssemblyAccumulator(lower)
|
||||||
|
if diagnostics := assembly.applyOverlayWithCollectionPolicy(higher.root, higher.origins, higher.collectionPolicies, higher.applications, LayerBase, policy, declarations); len(diagnostics) != 0 {
|
||||||
|
t.Fatal(diagnostics)
|
||||||
|
}
|
||||||
|
if len(assembly.current.applications) != 3 {
|
||||||
|
t.Fatalf("source applications = %d, want all three sources", len(assembly.current.applications))
|
||||||
|
}
|
||||||
|
if mode == sequenceModeUpsert {
|
||||||
|
query := Keys("records").Index(1).Key("value")
|
||||||
|
if _, relevant := applicationOperationPath(assembly.current.applications[1], query, nil); relevant {
|
||||||
|
t.Fatal("first peer was attributed to another peer's upserted record")
|
||||||
|
}
|
||||||
|
origin := historyApplicationOrigin(assembly.current.applications[2], query, nil)
|
||||||
|
if len(origin.Locations) != 1 || origin.Locations[0].SourceID != "second-peer" {
|
||||||
|
t.Fatalf("upsert history source = %#v", origin)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
working := parseCompositionDocument(t, "initial", "{}\n")
|
||||||
|
index := new(mappingIndex)
|
||||||
|
for _, application := range assembly.current.applications {
|
||||||
|
applyHistoryApplication(&working, application, index)
|
||||||
|
}
|
||||||
|
if !configurationValuesEqual(working, assembly.current.root) {
|
||||||
|
t.Fatal("tier history differs from structural assembly before decoding")
|
||||||
|
}
|
||||||
|
overlay := parseCompositionDocument(t, "overlay", "records: [{id: c, value: overlay}]\n")
|
||||||
|
declarations[0].mode = sequenceModeUpsert
|
||||||
|
if diagnostics := assembly.applyOverlayWithCollectionPolicy(overlay, nil, collectionPolicyScope{}, nil, LayerOverlay, policy, declarations); len(diagnostics) != 0 {
|
||||||
|
t.Fatal(diagnostics)
|
||||||
|
}
|
||||||
|
applyHistoryApplication(&working, assembly.current.applications[3], index)
|
||||||
|
if !configurationValuesEqual(working, assembly.current.root) {
|
||||||
|
t.Fatal("subsequent overlay history differs from structural assembly")
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestSequenceHistoryPreservesTierReplacementAndEmptyLists(t *testing.T) {
|
func TestSequenceHistoryPreservesTierReplacementAndEmptyLists(t *testing.T) {
|
||||||
policy := mustCompileSequenceIdentityPolicy(t, []sequenceIdentityRegistration{
|
policy := mustCompileSequenceIdentityPolicy(t, []sequenceIdentityRegistration{
|
||||||
{selector: sequenceIdentitySelector{literalSequenceSelector("records")}, identityFields: []string{"id"}},
|
{selector: sequenceIdentitySelector{literalSequenceSelector("records")}, identityFields: []string{"id"}},
|
||||||
@@ -151,6 +216,49 @@ func TestHistoryMergeSupportsSemanticIndexedPrefixes(t *testing.T) {
|
|||||||
if !present || value.Text != "new" {
|
if !present || value.Text != "new" {
|
||||||
t.Fatalf("indexed helper replay = %#v, %v", value, present)
|
t.Fatalf("indexed helper replay = %#v, %v", value, present)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
root = parseCompositionDocument(t, "base", "records: [{id: one, helper: {items: [{id: a}]}}]\n")
|
||||||
|
application.value = parseCompositionDocument(t, "helper", "items: [{id: b}]\n")
|
||||||
|
application.sequences = []sequenceHistoryContribution{{path: Keys("items"), mode: sequenceHistoryAppend, offset: 1, length: 1}}
|
||||||
|
application.sequencesByPath = indexSequenceHistoryContributions(application.sequences)
|
||||||
|
applyHistoryApplication(&root, application, new(mappingIndex))
|
||||||
|
items, present := documentAtPath(root, Keys("records").Index(0).Key("helper").Key("items"))
|
||||||
|
if !present || len(items.Sequence) != 2 {
|
||||||
|
t.Fatalf("indexed helper lost its local sequence operation: %#v", items)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestLayeredUpsertExplanationFollowsStructuralOrder(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
type config struct {
|
||||||
|
Records []map[string]any `yaml:"records"`
|
||||||
|
}
|
||||||
|
loaded, err := testGroupedLoad(loadRequest[config]{
|
||||||
|
main: loadInput{source: Bytes("main", []byte("configkit:\n sequences: [{destination: [records], key: id}]\nrecords: [{id: b, value: new-b}, {id: c, value: new-c}, {id: a, value: new-a}]\n"))},
|
||||||
|
underlays: []loadGroup{{inputs: []loadInput{{source: Bytes("lower", []byte("records: [{id: a, value: old-a}, {id: b, value: old-b, removed: old}]\n"))}}}},
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
explanation, ok := loaded.Explain(Keys("records"))
|
||||||
|
if !ok || len(explanation.History) == 0 {
|
||||||
|
t.Fatal("missing collection history")
|
||||||
|
}
|
||||||
|
for _, change := range explanation.History {
|
||||||
|
if change.Origin.Operation == ChangeDecoding {
|
||||||
|
t.Fatal("structural upsert was misreported as a decoding reconciliation")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if last := explanation.History[len(explanation.History)-1]; last.After.YAML != explanation.Effective.YAML || last.Origin.Layer != LayerBase {
|
||||||
|
t.Fatalf("main contribution did not produce the effective collection: %#v", last)
|
||||||
|
}
|
||||||
|
removed, ok := loaded.Explain(Keys("records").Index(1).Key("removed"))
|
||||||
|
if !ok || len(removed.History) == 0 || removed.Effective.Present {
|
||||||
|
t.Fatalf("missing displaced field history: %#v, %v", removed, ok)
|
||||||
|
}
|
||||||
|
if last := removed.History[len(removed.History)-1]; last.After.Present || last.Origin.Layer != LayerBase || last.Origin.Operation == ChangeDecoding {
|
||||||
|
t.Fatalf("upsert removal attribution = %#v", last)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestProjectedSequenceHistoryAgreesAcrossMetadataAndExplanation(t *testing.T) {
|
func TestProjectedSequenceHistoryAgreesAcrossMetadataAndExplanation(t *testing.T) {
|
||||||
|
|||||||
@@ -78,7 +78,7 @@ func (a *assemblyAccumulator) applyOverlayWithCollectionPolicy(
|
|||||||
application.sequencesByPath = indexSequenceHistoryContributions(application.sequences)
|
application.sequencesByPath = indexSequenceHistoryContributions(application.sequences)
|
||||||
a.current.applications = append(a.current.applications, application)
|
a.current.applications = append(a.current.applications, application)
|
||||||
} else {
|
} else {
|
||||||
for _, application := range applications {
|
for _, application := range projectTierHistoryApplications(applications, sequences) {
|
||||||
application.layer = layer
|
application.layer = layer
|
||||||
a.current.applications = append(a.current.applications, application)
|
a.current.applications = append(a.current.applications, application)
|
||||||
}
|
}
|
||||||
|
|||||||
76
internal/engine/sequence_history_projection.go
Normal file
76
internal/engine/sequence_history_projection.go
Normal file
@@ -0,0 +1,76 @@
|
|||||||
|
package engine
|
||||||
|
|
||||||
|
import "gitea.maximumdirect.net/eric/ConfigKit/internal/document"
|
||||||
|
|
||||||
|
// projectTierHistoryApplications composes each strict source's local projection
|
||||||
|
// with the tier operation. The retained source values stay in authored coordinates.
|
||||||
|
func projectTierHistoryApplications(applications []historyApplication, operations []sequenceHistoryContribution) []historyApplication {
|
||||||
|
mapped := false
|
||||||
|
for _, operation := range operations {
|
||||||
|
mapped = mapped || operation.mode == sequenceHistoryMapped
|
||||||
|
}
|
||||||
|
if !mapped {
|
||||||
|
return applications
|
||||||
|
}
|
||||||
|
projections := indexSequenceHistoryContributions(operations)
|
||||||
|
projected := cloneHistoryApplications(applications)
|
||||||
|
for index, original := range applications {
|
||||||
|
application := &projected[index]
|
||||||
|
if path, found := projectSequenceHistoryPath(original.path, projections); found {
|
||||||
|
application.path = path
|
||||||
|
}
|
||||||
|
for sequenceIndex, source := range original.sequences {
|
||||||
|
container := appendDestination(original.path, source.path)
|
||||||
|
operation, active := projections[container.String()]
|
||||||
|
if !active || operation.mode != sequenceHistoryMapped {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
value, found := documentAtPath(original.value, source.path)
|
||||||
|
if !found || value.Kind != document.Sequence {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
sourceIndex := application.sequencesByPath[source.path.String()]
|
||||||
|
contribution := sequenceHistoryContribution{path: source.path.clone(), mode: sequenceHistoryMapped}
|
||||||
|
for inputIndex := range value.Sequence {
|
||||||
|
tierIndex := source.offset + inputIndex
|
||||||
|
if source.mode == sequenceHistoryMapped {
|
||||||
|
tierIndex = sourceIndex.outputForInput(inputIndex)
|
||||||
|
}
|
||||||
|
outputIndex, present := operation.projection.outputForInput(tierIndex)
|
||||||
|
if !present {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
contribution.items = append(contribution.items, sequenceItemProjection{inputIndex: inputIndex, outputIndex: outputIndex})
|
||||||
|
if _, replaced := operation.replacedOutputs[outputIndex]; replaced {
|
||||||
|
// Upsert replacements retain their lower-list position.
|
||||||
|
contribution.replacements = append(contribution.replacements, sequenceItemReplacement{
|
||||||
|
lowerIndex: outputIndex, higherIndex: inputIndex, outputIndex: outputIndex,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
application.sequences[sequenceIndex] = contribution
|
||||||
|
}
|
||||||
|
application.sequencesByPath = indexSequenceHistoryContributions(application.sequences)
|
||||||
|
}
|
||||||
|
return projected
|
||||||
|
}
|
||||||
|
|
||||||
|
func projectSequenceHistoryPath(path Path, projections map[string]sequenceHistoryContributionIndex) (Path, bool) {
|
||||||
|
projected := path.clone()
|
||||||
|
for index, segment := range path.segments {
|
||||||
|
if segment.kind != pathSegmentIndex {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
prefix := Path{segments: path.segments[:index]}
|
||||||
|
operation, found := projections[prefix.String()]
|
||||||
|
if !found || operation.mode != sequenceHistoryMapped {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
outputIndex, found := operation.projection.outputForInput(segment.index)
|
||||||
|
if !found {
|
||||||
|
return Path{}, false
|
||||||
|
}
|
||||||
|
projected.segments[index].index = outputIndex
|
||||||
|
}
|
||||||
|
return projected, true
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user