253 lines
11 KiB
Go
253 lines
11 KiB
Go
package checkpoint
|
|
|
|
import (
|
|
"encoding/base64"
|
|
"encoding/json"
|
|
"fmt"
|
|
"os"
|
|
"strings"
|
|
|
|
"gitea.maximumdirect.net/eric/notarius/internal/core/fileio"
|
|
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
|
|
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
|
|
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
|
|
)
|
|
|
|
type FilesystemLoader struct {
|
|
root string
|
|
identityDigest string
|
|
}
|
|
|
|
func NewFilesystemLoader(root string, identity Identity) (pipeline.CheckpointLoader, error) {
|
|
root = strings.TrimSpace(root)
|
|
if root == "" {
|
|
return pipeline.NoopCheckpointLoader(), nil
|
|
}
|
|
relative, err := identity.RelativePath()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
target, err := fileio.SafePath(root, relative)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &FilesystemLoader{root: target, identityDigest: identity.Digest}, nil
|
|
}
|
|
|
|
func (l *FilesystemLoader) Enabled() bool {
|
|
return l != nil && strings.TrimSpace(l.root) != ""
|
|
}
|
|
|
|
func (l *FilesystemLoader) Source(moduleKey string) (pipeline.SourceCheckpoint, pipeline.CheckpointDecision) {
|
|
var manifest SourceManifest
|
|
if decision := l.readJSON("source/manifest.json", &manifest); !decision.Reused {
|
|
return pipeline.SourceCheckpoint{}, decision
|
|
}
|
|
if decision := l.validateManifest(manifest.StageManifest, StageSource, "", moduleKey, StatusSucceeded, nil); !decision.Reused {
|
|
return pipeline.SourceCheckpoint{}, decision
|
|
}
|
|
var payload sourceDocumentEnvelope
|
|
if decision := l.readJSON("source/source-document.json", &payload); !decision.Reused {
|
|
return pipeline.SourceCheckpoint{}, decision
|
|
}
|
|
doc := cloneSourceDocument(payload.Document)
|
|
if err := source.ValidateDocument(&doc); err != nil {
|
|
return pipeline.SourceCheckpoint{}, invalidDecision("source checkpoint document is invalid: %v", err)
|
|
}
|
|
if strings.TrimSpace(manifest.SourceID) != "" && manifest.SourceID != doc.ID {
|
|
return pipeline.SourceCheckpoint{}, invalidDecision("source checkpoint source id does not match payload")
|
|
}
|
|
if !fingerprintsEqual(checkpointToPipelineFingerprints(manifest.OutputDigests), digestFingerprints("source_document", doc.Digest)) {
|
|
return pipeline.SourceCheckpoint{}, invalidDecision("source checkpoint output digest does not match payload")
|
|
}
|
|
return pipeline.SourceCheckpoint{Document: &doc}, reusedDecision()
|
|
}
|
|
|
|
func (l *FilesystemLoader) Extract(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.ExtractCheckpoint, pipeline.CheckpointDecision) {
|
|
var manifest ExtractLaneManifest
|
|
if d := l.readJSON(laneManifestPath("extract", laneID), &manifest); !d.Reused {
|
|
return pipeline.ExtractCheckpoint{}, d
|
|
}
|
|
if d := l.validateLaneManifest(manifest.StageManifest, StageExtract, laneID, moduleKey, dependencies, StatusSucceeded, StatusSucceededWithRejections); !d.Reused {
|
|
return pipeline.ExtractCheckpoint{}, d
|
|
}
|
|
var payload artifactExtractEnvelope
|
|
if d := l.readJSON(lanePayloadPath("extract", laneID, "outputs.json"), &payload); !d.Reused {
|
|
return pipeline.ExtractCheckpoint{}, d
|
|
}
|
|
outputs, err := artifactCheckpointOutputs(payload.Outputs)
|
|
if err != nil {
|
|
return pipeline.ExtractCheckpoint{}, invalidDecision("extract artifact checkpoint payload is invalid: %v", err)
|
|
}
|
|
if !fingerprintsEqual(checkpointToPipelineFingerprints(manifest.OutputDigests), artifactOutputDigests(outputs)) {
|
|
return pipeline.ExtractCheckpoint{}, invalidDecision("extract artifact checkpoint output digests do not match payload")
|
|
}
|
|
return pipeline.ExtractCheckpoint{Outputs: outputs, Rejected: cloneRejectedOutputs(payload.Rejected), Warnings: cloneWarnings(payload.Warnings)}, reusedDecision()
|
|
}
|
|
|
|
func (l *FilesystemLoader) Merge(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.MergeCheckpoint, pipeline.CheckpointDecision) {
|
|
var manifest MergeLaneManifest
|
|
if d := l.readJSON(laneManifestPath("merge", laneID), &manifest); !d.Reused {
|
|
return pipeline.MergeCheckpoint{}, d
|
|
}
|
|
if d := l.validateLaneManifest(manifest.StageManifest, StageMerge, laneID, moduleKey, dependencies, StatusSucceeded); !d.Reused {
|
|
return pipeline.MergeCheckpoint{}, d
|
|
}
|
|
var payload artifactSingleEnvelope
|
|
if d := l.readJSON(lanePayloadPath("merge", laneID, "output.json"), &payload); !d.Reused {
|
|
return pipeline.MergeCheckpoint{}, d
|
|
}
|
|
values, err := artifactCheckpointOutputs([]artifactCheckpointEnvelope{payload.Output})
|
|
if err != nil {
|
|
return pipeline.MergeCheckpoint{}, invalidDecision("merge artifact checkpoint payload is invalid: %v", err)
|
|
}
|
|
if !fingerprintsEqual(checkpointToPipelineFingerprints(manifest.OutputDigests), artifactOutputDigests(values)) {
|
|
return pipeline.MergeCheckpoint{}, invalidDecision("merge artifact checkpoint output digest does not match payload")
|
|
}
|
|
return pipeline.MergeCheckpoint{Output: values[0], Warnings: cloneWarnings(payload.Warnings)}, reusedDecision()
|
|
}
|
|
|
|
func (l *FilesystemLoader) Normalize(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.NormalizeCheckpoint, pipeline.CheckpointDecision) {
|
|
var manifest NormalizeLaneManifest
|
|
if d := l.readJSON(laneManifestPath("normalize", laneID), &manifest); !d.Reused {
|
|
return pipeline.NormalizeCheckpoint{}, d
|
|
}
|
|
if d := l.validateLaneManifest(manifest.StageManifest, StageNormalize, laneID, moduleKey, dependencies, StatusSucceeded); !d.Reused {
|
|
return pipeline.NormalizeCheckpoint{}, d
|
|
}
|
|
var payload artifactSingleEnvelope
|
|
if d := l.readJSON(lanePayloadPath("normalize", laneID, "output.json"), &payload); !d.Reused {
|
|
return pipeline.NormalizeCheckpoint{}, d
|
|
}
|
|
values, err := artifactCheckpointOutputs([]artifactCheckpointEnvelope{payload.Output})
|
|
if err != nil {
|
|
return pipeline.NormalizeCheckpoint{}, invalidDecision("normalize artifact checkpoint payload is invalid: %v", err)
|
|
}
|
|
if !fingerprintsEqual(checkpointToPipelineFingerprints(manifest.OutputDigests), artifactOutputDigests(values)) {
|
|
return pipeline.NormalizeCheckpoint{}, invalidDecision("normalize artifact checkpoint output digest does not match payload")
|
|
}
|
|
return pipeline.NormalizeCheckpoint{Output: values[0], Warnings: cloneWarnings(payload.Warnings)}, reusedDecision()
|
|
}
|
|
|
|
func artifactCheckpointOutputs(values []artifactCheckpointEnvelope) ([]pipeline.CheckpointArtifact, error) {
|
|
if len(values) == 0 {
|
|
return nil, nil
|
|
}
|
|
out := make([]pipeline.CheckpointArtifact, 0, len(values))
|
|
for _, v := range values {
|
|
content, err := contentFromEnvelope(v.Content)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if strings.TrimSpace(string(v.Kind)) == "" || strings.TrimSpace(v.Schema.ID) == "" || strings.TrimSpace(v.Schema.Version) == "" || strings.TrimSpace(v.SchemaDigest) == "" {
|
|
return nil, fmt.Errorf("artifact codec identity is incomplete")
|
|
}
|
|
out = append(out, pipeline.CheckpointArtifact{LaneID: v.LaneID, ModuleKey: v.ModuleKey, SourceID: v.SourceID, ChunkID: v.ChunkID, ChunkIndex: v.ChunkIndex, ChunkRef: v.ChunkRef, SchemaDigest: v.SchemaDigest, Artifact: contracts.SerializedArtifact{Kind: v.Kind, Schema: v.Schema, MediaType: v.Content.MediaType, Content: content, Metadata: cloneMetadata(v.Content.Metadata)}})
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func (l *FilesystemLoader) readJSON(name string, out any) pipeline.CheckpointDecision {
|
|
if !l.Enabled() {
|
|
return pipeline.CheckpointDecision{Reason: "checkpoint loading disabled"}
|
|
}
|
|
target, err := fileio.SafePath(l.root, name)
|
|
if err != nil {
|
|
return invalidDecision("checkpoint path is invalid: %v", err)
|
|
}
|
|
data, err := os.ReadFile(target)
|
|
if err != nil {
|
|
if os.IsNotExist(err) {
|
|
return pipeline.CheckpointDecision{Reason: "checkpoint artifact is missing"}
|
|
}
|
|
return invalidDecision("read checkpoint artifact: %v", err)
|
|
}
|
|
if err := json.Unmarshal(data, out); err != nil {
|
|
return invalidDecision("decode checkpoint artifact: %v", err)
|
|
}
|
|
return reusedDecision()
|
|
}
|
|
|
|
func (l *FilesystemLoader) validateManifest(manifest StageManifest, stage StageName, laneID string, moduleKey string, status StageStatus, dependencies []pipeline.CheckpointFingerprint) pipeline.CheckpointDecision {
|
|
return l.validateLaneManifest(manifest, stage, laneID, moduleKey, dependencies, status)
|
|
}
|
|
|
|
func (l *FilesystemLoader) validateLaneManifest(manifest StageManifest, stage StageName, laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, statuses ...StageStatus) pipeline.CheckpointDecision {
|
|
if manifest.WorkspaceSchemaVersion == WorkspaceSchemaVersionV1 {
|
|
return invalidDecision("checkpoint workspace schema version %q is incompatible with %q and must be recomputed", manifest.WorkspaceSchemaVersion, WorkspaceSchemaVersion)
|
|
}
|
|
if manifest.WorkspaceSchemaVersion != WorkspaceSchemaVersion {
|
|
return invalidDecision("checkpoint workspace schema version %q is not supported", manifest.WorkspaceSchemaVersion)
|
|
}
|
|
if strings.TrimSpace(l.identityDigest) != "" && manifest.Metadata["checkpoint_identity_digest"] != l.identityDigest {
|
|
return invalidDecision("checkpoint identity digest does not match current invocation")
|
|
}
|
|
if manifest.Stage != stage {
|
|
return invalidDecision("checkpoint stage %q does not match %q", manifest.Stage, stage)
|
|
}
|
|
if strings.TrimSpace(laneID) != "" && manifest.LaneID != laneID {
|
|
return invalidDecision("checkpoint lane %q does not match %q", manifest.LaneID, laneID)
|
|
}
|
|
if strings.TrimSpace(moduleKey) != "" && manifest.ModuleKey != moduleKey {
|
|
return invalidDecision("checkpoint module %q does not match %q", manifest.ModuleKey, moduleKey)
|
|
}
|
|
statusOK := false
|
|
for _, status := range statuses {
|
|
if manifest.Status == status {
|
|
statusOK = true
|
|
break
|
|
}
|
|
}
|
|
if !statusOK {
|
|
return invalidDecision("checkpoint status %q cannot be reused", manifest.Status)
|
|
}
|
|
if !fingerprintsEqual(checkpointToPipelineFingerprints(manifest.DependencyFingerprints), dependencies) {
|
|
return invalidDecision("checkpoint dependency fingerprints do not match")
|
|
}
|
|
return reusedDecision()
|
|
}
|
|
|
|
func contentFromEnvelope(value binaryEnvelope) ([]byte, error) {
|
|
content, err := base64.StdEncoding.DecodeString(value.ContentBase64)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("decode content_base64: %w", err)
|
|
}
|
|
if digest := strings.TrimSpace(value.ContentDigest); digest != "" && digest != contentDigest(content) {
|
|
return nil, fmt.Errorf("content digest mismatch")
|
|
}
|
|
return content, nil
|
|
}
|
|
|
|
func checkpointToPipelineFingerprints(values []Fingerprint) []pipeline.CheckpointFingerprint {
|
|
if len(values) == 0 {
|
|
return nil
|
|
}
|
|
out := make([]pipeline.CheckpointFingerprint, 0, len(values))
|
|
for _, value := range values {
|
|
out = append(out, pipeline.CheckpointFingerprint{Name: value.Name, Value: value.Value})
|
|
}
|
|
return normalizeFingerprints(out)
|
|
}
|
|
|
|
func fingerprintsEqual(a []pipeline.CheckpointFingerprint, b []pipeline.CheckpointFingerprint) bool {
|
|
a = normalizeFingerprints(a)
|
|
b = normalizeFingerprints(b)
|
|
if len(a) != len(b) {
|
|
return false
|
|
}
|
|
for i := range a {
|
|
if a[i] != b[i] {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
func reusedDecision() pipeline.CheckpointDecision {
|
|
return pipeline.CheckpointDecision{Reused: true, Reason: "checkpoint is valid"}
|
|
}
|
|
|
|
func invalidDecision(format string, args ...any) pipeline.CheckpointDecision {
|
|
return pipeline.CheckpointDecision{Reason: fmt.Sprintf(format, args...)}
|
|
}
|