100 lines
3.3 KiB
Go
100 lines
3.3 KiB
Go
package stage
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
|
|
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
|
)
|
|
|
|
// checkpointStageCollection isolates one stage-owned collection update from
|
|
// live authority until the existing manifest store reports a successful save.
|
|
// Callers own serialization and all stage-specific validation and policy.
|
|
func checkpointStageCollection(
|
|
ctx context.Context,
|
|
store manifest.Store,
|
|
path string,
|
|
live *manifest.Manifest,
|
|
stageName string,
|
|
update func(*manifest.StageRecord) error,
|
|
) error {
|
|
if live == nil {
|
|
return fmt.Errorf("session manifest is required")
|
|
}
|
|
if store == nil {
|
|
return fmt.Errorf("manifest store is required")
|
|
}
|
|
if update == nil {
|
|
return fmt.Errorf("checkpoint update is required")
|
|
}
|
|
|
|
candidate := cloneManifestForCheckpoint(live)
|
|
if candidate.Stages == nil {
|
|
candidate.Stages = make(map[string]*manifest.StageRecord)
|
|
}
|
|
target, exists := candidate.Stages[stageName]
|
|
if !exists {
|
|
target = &manifest.StageRecord{}
|
|
candidate.Stages[stageName] = target
|
|
} else if target == nil {
|
|
// Retain the invalid nil entry for Store.Save to reject rather than
|
|
// silently repairing manifest state. The detached record only gives the
|
|
// typed update a safe value to operate on before that validation.
|
|
target = &manifest.StageRecord{}
|
|
}
|
|
if err := update(target); err != nil {
|
|
return fmt.Errorf("update stage %q checkpoint: %w", stageName, err)
|
|
}
|
|
if err := store.Save(ctx, path, candidate); err != nil {
|
|
return fmt.Errorf("save session manifest %q: %w", path, err)
|
|
}
|
|
|
|
savedRecord := cloneStageRecordForCheckpoint(candidate.Stages[stageName])
|
|
if savedRecord == nil {
|
|
return fmt.Errorf("saved stage %q checkpoint record is nil", stageName)
|
|
}
|
|
if live.Stages != nil {
|
|
if current := live.Stages[stageName]; current != nil {
|
|
*current = *savedRecord
|
|
} else {
|
|
live.Stages[stageName] = savedRecord
|
|
}
|
|
} else {
|
|
live.Stages = map[string]*manifest.StageRecord{stageName: savedRecord}
|
|
}
|
|
live.UpdatedAt = candidate.UpdatedAt
|
|
return nil
|
|
}
|
|
|
|
// cloneManifestForCheckpoint copies the bounded mutable state touched by the
|
|
// two checkpoint callbacks or by manifest.LocalStore.Save. Metadata maps and
|
|
// other pointer fields remain shared read-only because neither boundary mutates
|
|
// them.
|
|
func cloneManifestForCheckpoint(source *manifest.Manifest) *manifest.Manifest {
|
|
clone := *source
|
|
clone.Inputs = append([]manifest.InputRecord(nil), source.Inputs...)
|
|
clone.Artifacts = append([]manifest.ArtifactRecord(nil), source.Artifacts...)
|
|
if source.Stages == nil {
|
|
clone.Stages = nil
|
|
return &clone
|
|
}
|
|
clone.Stages = make(map[string]*manifest.StageRecord, len(source.Stages))
|
|
for name, sourceRecord := range source.Stages {
|
|
clone.Stages[name] = cloneStageRecordForCheckpoint(sourceRecord)
|
|
}
|
|
return &clone
|
|
}
|
|
|
|
func cloneStageRecordForCheckpoint(source *manifest.StageRecord) *manifest.StageRecord {
|
|
if source == nil {
|
|
return nil
|
|
}
|
|
clone := *source
|
|
clone.Outputs = append([]manifest.ArtifactRecord(nil), source.Outputs...)
|
|
clone.Logs = append([]string(nil), source.Logs...)
|
|
clone.GeneratedConfigs = append([]string(nil), source.GeneratedConfigs...)
|
|
clone.TranscribeSpeakers = manifest.CloneTranscribeSpeakerCollection(source.TranscribeSpeakers)
|
|
clone.AnalyzeArtifacts = manifest.CloneAnalyzeArtifactCollection(source.AnalyzeArtifacts)
|
|
return &clone
|
|
}
|