eric/notarius
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Decouple checkpoint storage from workspace state

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-18 04:55:33 +00:00
parent 286fb9dce7
commit 5bd0ba7a72
17 changed files with 497 additions and 1135 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
internal/framework/checkpoint/loader.go
View File

@@ -7,38 +7,43 @@ import (
"os"
"strings"
"gitea.maximumdirect.net/eric/notarius/internal/core/fileio"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
coreworkspace "gitea.maximumdirect.net/eric/notarius/internal/core/workspace"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
)
type WorkspaceLoader struct {
type FilesystemLoader struct {
root string
identityDigest string
}
func NewWorkspaceLoader(settings coreworkspace.Settings, identity coreworkspace.CheckpointIdentity) (pipeline.CheckpointLoader, error) {
root, err := settings.CheckpointDirectory(identity)
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
}
if strings.TrimSpace(root) == "" {
return pipeline.NoopCheckpointLoader(), nil
target, err := fileio.SafePath(root, relative)
if err != nil {
return nil, err
}
return &WorkspaceLoader{root: root, identityDigest: identity.Digest}, nil
return &FilesystemLoader{root: target, identityDigest: identity.Digest}, nil
}
func (l *WorkspaceLoader) Enabled() bool {
func (l *FilesystemLoader) Enabled() bool {
return l != nil && strings.TrimSpace(l.root) != ""
}
func (l *WorkspaceLoader) Source(moduleKey string) (pipeline.SourceCheckpoint, pipeline.CheckpointDecision) {
var manifest coreworkspace.SourceManifest
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, coreworkspace.StageSource, "", moduleKey, coreworkspace.StatusSucceeded, nil); !decision.Reused {
if decision := l.validateManifest(manifest.StageManifest, StageSource, "", moduleKey, StatusSucceeded, nil); !decision.Reused {
return pipeline.SourceCheckpoint{}, decision
}
var payload sourceDocumentEnvelope
@@ -52,18 +57,18 @@ func (l *WorkspaceLoader) Source(moduleKey string) (pipeline.SourceCheckpoint, p
if strings.TrimSpace(manifest.SourceID) != "" && manifest.SourceID != doc.ID {
return pipeline.SourceCheckpoint{}, invalidDecision("source checkpoint source id does not match payload")
}
if !fingerprintsEqual(coreworkspaceToPipelineFingerprints(manifest.OutputDigests), digestFingerprints("source_document", doc.Digest)) {
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 *WorkspaceLoader) Extract(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.ExtractCheckpoint, pipeline.CheckpointDecision) {
var manifest coreworkspace.ExtractLaneManifest
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, coreworkspace.StageExtract, laneID, moduleKey, dependencies, coreworkspace.StatusSucceeded, coreworkspace.StatusSucceededWithRejections); !d.Reused {
if d := l.validateLaneManifest(manifest.StageManifest, StageExtract, laneID, moduleKey, dependencies, StatusSucceeded, StatusSucceededWithRejections); !d.Reused {
return pipeline.ExtractCheckpoint{}, d
}
var payload artifactExtractEnvelope
@@ -74,18 +79,18 @@ func (l *WorkspaceLoader) Extract(laneID, moduleKey string, dependencies []pipel
if err != nil {
return pipeline.ExtractCheckpoint{}, invalidDecision("extract artifact checkpoint payload is invalid: %v", err)
}
if !fingerprintsEqual(coreworkspaceToPipelineFingerprints(manifest.OutputDigests), artifactOutputDigests(outputs)) {
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 *WorkspaceLoader) Merge(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.MergeCheckpoint, pipeline.CheckpointDecision) {
var manifest coreworkspace.MergeLaneManifest
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, coreworkspace.StageMerge, laneID, moduleKey, dependencies, coreworkspace.StatusSucceeded); !d.Reused {
if d := l.validateLaneManifest(manifest.StageManifest, StageMerge, laneID, moduleKey, dependencies, StatusSucceeded); !d.Reused {
return pipeline.MergeCheckpoint{}, d
}
var payload artifactSingleEnvelope
@@ -96,18 +101,18 @@ func (l *WorkspaceLoader) Merge(laneID, moduleKey string, dependencies []pipelin
if err != nil {
return pipeline.MergeCheckpoint{}, invalidDecision("merge artifact checkpoint payload is invalid: %v", err)
}
if !fingerprintsEqual(coreworkspaceToPipelineFingerprints(manifest.OutputDigests), artifactOutputDigests(values)) {
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 *WorkspaceLoader) Normalize(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.NormalizeCheckpoint, pipeline.CheckpointDecision) {
var manifest coreworkspace.NormalizeLaneManifest
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, coreworkspace.StageNormalize, laneID, moduleKey, dependencies, coreworkspace.StatusSucceeded); !d.Reused {
if d := l.validateLaneManifest(manifest.StageManifest, StageNormalize, laneID, moduleKey, dependencies, StatusSucceeded); !d.Reused {
return pipeline.NormalizeCheckpoint{}, d
}
var payload artifactSingleEnvelope
@@ -118,7 +123,7 @@ func (l *WorkspaceLoader) Normalize(laneID, moduleKey string, dependencies []pip
if err != nil {
return pipeline.NormalizeCheckpoint{}, invalidDecision("normalize artifact checkpoint payload is invalid: %v", err)
}
if !fingerprintsEqual(coreworkspaceToPipelineFingerprints(manifest.OutputDigests), artifactOutputDigests(values)) {
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()
@@ -142,11 +147,11 @@ func artifactCheckpointOutputs(values []artifactCheckpointEnvelope) ([]pipeline.
return out, nil
}
func (l *WorkspaceLoader) readJSON(name string, out any) pipeline.CheckpointDecision {
func (l *FilesystemLoader) readJSON(name string, out any) pipeline.CheckpointDecision {
if !l.Enabled() {
return pipeline.CheckpointDecision{Reason: "checkpoint loading disabled"}
}
target, err := coreworkspace.SafePath(l.root, name)
target, err := fileio.SafePath(l.root, name)
if err != nil {
return invalidDecision("checkpoint path is invalid: %v", err)
}
@@ -163,15 +168,15 @@ func (l *WorkspaceLoader) readJSON(name string, out any) pipeline.CheckpointDeci
return reusedDecision()
}
func (l *WorkspaceLoader) validateManifest(manifest coreworkspace.StageManifest, stage coreworkspace.StageName, laneID string, moduleKey string, status coreworkspace.StageStatus, dependencies []pipeline.CheckpointFingerprint) pipeline.CheckpointDecision {
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 *WorkspaceLoader) validateLaneManifest(manifest coreworkspace.StageManifest, stage coreworkspace.StageName, laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, statuses ...coreworkspace.StageStatus) pipeline.CheckpointDecision {
if manifest.WorkspaceSchemaVersion == coreworkspace.WorkspaceSchemaVersionV1 {
return invalidDecision("checkpoint workspace schema version %q is incompatible with %q and must be recomputed", manifest.WorkspaceSchemaVersion, coreworkspace.WorkspaceSchemaVersion)
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 != coreworkspace.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 {
@@ -196,7 +201,7 @@ func (l *WorkspaceLoader) validateLaneManifest(manifest coreworkspace.StageManif
if !statusOK {
return invalidDecision("checkpoint status %q cannot be reused", manifest.Status)
}
if !fingerprintsEqual(coreworkspaceToPipelineFingerprints(manifest.DependencyFingerprints), dependencies) {
if !fingerprintsEqual(checkpointToPipelineFingerprints(manifest.DependencyFingerprints), dependencies) {
return invalidDecision("checkpoint dependency fingerprints do not match")
}
return reusedDecision()
@@ -213,7 +218,7 @@ func contentFromEnvelope(value binaryEnvelope) ([]byte, error) {
return content, nil
}
func coreworkspaceToPipelineFingerprints(values []coreworkspace.Fingerprint) []pipeline.CheckpointFingerprint {
func checkpointToPipelineFingerprints(values []Fingerprint) []pipeline.CheckpointFingerprint {
if len(values) == 0 {
return nil
}