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

Add reconciliation state foundation

Browse Source
This commit is contained in:
Eric Rakestraw
2026-06-08 18:24:12 +00:00
parent c04432e40b
commit b7db3993fb
13 changed files with 557 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
internal/config/config.go
View File

@@ -55,6 +55,7 @@ type Destination struct {
Transform Transform `yaml:"transform"`
PathMap PathMapping `yaml:"path_mapping"`
Links *Links `yaml:"links"`
Reconciliation ReconciliationPolicy `yaml:"reconciliation"`
Transfer TransferPolicy `yaml:"transfer"`
}
@@ -118,6 +119,10 @@ type Links struct {
Primary string `yaml:"primary"`
}
type ReconciliationPolicy struct {
Mode string `yaml:"mode"`
}
type TransferPolicy struct {
OnDestinationSame string `yaml:"on_destination_same"`
OnDestinationOlder string `yaml:"on_destination_older"`

8
internal/config/defaults.go
View File

@@ -42,6 +42,11 @@ const (
LinkPrimarySource = "source"
)
const (
ReconciliationModeReplace = "replace"
ReconciliationModeMerge = "merge"
)
const DefaultS3Region = "us-east-1"
const (
@@ -79,6 +84,9 @@ func ApplyDefaults(cfg *Config) {
if destination.Links != nil && destination.Links.Primary == "" {
destination.Links.Primary = LinkPrimaryAuto
}
if destination.Reconciliation.Mode == "" {
destination.Reconciliation.Mode = ReconciliationModeReplace
}
if destination.Transfer.OnDestinationSame == "" {
destination.Transfer.OnDestinationSame = TransferActionSkip
}

32
internal/config/load_test.go
View File

@@ -33,6 +33,9 @@ pipelines:
if got, want := destination.Transfer.OnDestinationOlder, TransferActionReplace; got != want {
t.Fatalf("transfer default = %q, want %q", got, want)
}
if got, want := destination.Reconciliation.Mode, ReconciliationModeReplace; got != want {
t.Fatalf("reconciliation mode default = %q, want %q", got, want)
}
if cfg.Secrets.Directory != "" {
t.Fatalf("secrets.directory = %q, want empty", cfg.Secrets.Directory)
}
@@ -164,6 +167,35 @@ pipelines:
}
}
func TestLoadFileAcceptsExplicitReconciliationModes(t *testing.T) {
cfg := loadConfig(t, `
pipelines:
- id: reports
source:
backend: local
path: /source
destinations:
- id: archive
backend: local
path: /archive
reconciliation:
mode: replace
- id: web
backend: local
path: /web
reconciliation:
mode: merge
`)
destinations := cfg.Pipelines[0].Destinations
if got, want := destinations[0].Reconciliation.Mode, ReconciliationModeReplace; got != want {
t.Fatalf("archive reconciliation mode = %q, want %q", got, want)
}
if got, want := destinations[1].Reconciliation.Mode, ReconciliationModeMerge; got != want {
t.Fatalf("web reconciliation mode = %q, want %q", got, want)
}
}
func TestLoadFileAcceptsFixedPathMapping(t *testing.T) {
cfg := loadConfig(t, `
pipelines:

8
internal/config/validate.go
View File

@@ -72,6 +72,7 @@ func Validate(cfg Config) error {
errs = validatePublishTransformPolicy(errs, destinationContext, destination.Publish, destination.Transform)
errs = validatePathMapping(errs, destinationContext+".path_mapping", destination.PathMap)
errs = validateLinks(errs, destinationContext+".links", destination.Links)
errs = validateReconciliationPolicy(errs, destinationContext+".reconciliation", destination.Reconciliation)
errs = validateTransferPolicy(errs, destinationContext+".transfer", destination.Transfer)
}
}
@@ -314,6 +315,13 @@ func validateLinks(errs ValidationErrors, context string, links *Links) Validati
return errs
}
func validateReconciliationPolicy(errs ValidationErrors, context string, policy ReconciliationPolicy) ValidationErrors {
if policy.Mode != ReconciliationModeReplace && policy.Mode != ReconciliationModeMerge {
errs = append(errs, context+".mode must be "+ReconciliationModeReplace+" or "+ReconciliationModeMerge)
}
return errs
}
func validateTransferPolicy(errs ValidationErrors, context string, policy TransferPolicy) ValidationErrors {
if policy.OnDestinationSame != TransferActionSkip && policy.OnDestinationSame != TransferActionFail {
errs = append(errs, context+".on_destination_same must be skip or fail")

34
internal/config/validate_test.go
View File

@@ -107,6 +107,40 @@ func TestValidatePathMapping(t *testing.T) {
}
}
func TestValidateReconciliationPolicy(t *testing.T) {
tests := []struct {
name string
mode string
wantErr bool
}{
{name: "replace", mode: ReconciliationModeReplace},
{name: "merge", mode: ReconciliationModeMerge},
{name: "invalid", mode: "append", wantErr: true},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
cfg := Config{Pipelines: []Pipeline{{
ID: "reports",
Source: Backend{Backend: BackendLocal, Path: "/source"},
Destinations: []Destination{{
ID: "archive",
Backend: BackendLocal,
Path: "/destination",
Reconciliation: ReconciliationPolicy{Mode: tt.mode},
}},
}}}
ApplyDefaults(&cfg)
err := Validate(cfg)
if tt.wantErr && err == nil {
t.Fatal("Validate() error = nil, want error")
}
if !tt.wantErr && err != nil {
t.Fatalf("Validate() error = %v", err)
}
})
}
}
func TestValidateLinks(t *testing.T) {
tests := []struct {
name string

16
internal/publish/execute.go
View File

@@ -6,6 +6,7 @@ import (
"fmt"
"time"
"gitea.maximumdirect.net/eric/distributor/internal/config"
"gitea.maximumdirect.net/eric/distributor/internal/state"
"gitea.maximumdirect.net/eric/distributor/internal/storage"
)
@@ -69,14 +70,25 @@ func Execute(ctx context.Context, req Request, plan Plan) error {
writtenOutputs = append(writtenOutputs, output)
}
now := time.Now().UTC()
createdAt := now
existingOutputs := []state.OutputFile(nil)
if plan.ExistingState != nil {
createdAt = plan.ExistingState.CreatedAt
existingOutputs = plan.ExistingState.Outputs
}
destinationState := state.DistributorState{
SchemaVersion: state.SchemaVersion,
DistributorVersion: req.DistributorVersion,
PipelineID: req.PipelineID,
DestinationID: req.DestinationID,
PublishedAt: time.Now().UTC(),
PublishedAt: now,
CreatedAt: createdAt,
UpdatedAt: now,
State: state.StatePolicy{Mode: state.StateModeSingleOwner},
Reconciliation: state.ReconciliationPolicy{Mode: config.ReconciliationModeReplace},
Source: state.SourceState{Manifest: req.SourceBundle.Manifest},
Outputs: StateOutputFiles(plan.Outputs),
Outputs: state.ProjectOutputs(StateOutputProjections(plan.Outputs), existingOutputs, now),
}
if plan.PrimaryURL != "" {
destinationState.Links = &state.LinkState{PrimaryURL: plan.PrimaryURL}

20
internal/publish/output.go
View File

@@ -109,6 +109,18 @@ func (o Output) StateOutputFile() state.OutputFile {
}
}
func (o Output) StateOutputProjection() state.OutputProjection {
return state.OutputProjection{
Path: o.DestinationPath,
Kind: o.Kind,
SourcePath: o.SourcePath,
Transform: o.Transform,
URL: o.URL,
SHA256: o.SHA256,
Size: o.Size,
}
}
func (o Output) ManagedPath() string {
return o.DestinationPath
}
@@ -121,6 +133,14 @@ func StateOutputFiles(outputs []Output) []state.OutputFile {
return files
}
func StateOutputProjections(outputs []Output) []state.OutputProjection {
projections := make([]state.OutputProjection, 0, len(outputs))
for _, output := range outputs {
projections = append(projections, output.StateOutputProjection())
}
return projections
}
func ManagedOutputPaths(outputs []Output) []string {
paths := make([]string, 0, len(outputs))
for _, output := range outputs {

10
internal/state/compare_test.go
View File

@@ -6,6 +6,7 @@ import (
"time"
"gitea.maximumdirect.net/eric/distributor/internal/bundle"
"gitea.maximumdirect.net/eric/distributor/internal/config"
)
func TestCompareOutcomes(t *testing.T) {
@@ -100,11 +101,16 @@ func withState(t *testing.T, source bundle.Manifest, mutate func(*DistributorSta
t.Helper()
stateManifest := source
stateManifest.Files = append([]bundle.ManifestFile(nil), source.Files...)
publishedAt := time.Date(2026, 5, 30, 11, 12, 0, 0, time.UTC)
state := DistributorState{
SchemaVersion: SchemaVersion,
PipelineID: "reports",
DestinationID: "archive",
PublishedAt: time.Date(2026, 5, 30, 11, 12, 0, 0, time.UTC),
PublishedAt: publishedAt,
CreatedAt: publishedAt,
UpdatedAt: publishedAt,
State: StatePolicy{Mode: StateModeSingleOwner},
Reconciliation: ReconciliationPolicy{Mode: config.ReconciliationModeReplace},
Source: SourceState{Manifest: stateManifest},
Outputs: []OutputFile{{
Path: "report.md",
@@ -112,6 +118,8 @@ func withState(t *testing.T, source bundle.Manifest, mutate func(*DistributorSta
SourcePath: "report.md",
SHA256: source.Files[0].SHA256,
Size: source.Files[0].Size,
CreatedAt: publishedAt,
UpdatedAt: publishedAt,
}},
}
if mutate != nil {

142
internal/state/distributor.go
View File

@@ -8,9 +8,14 @@ import (
"time"
"gitea.maximumdirect.net/eric/distributor/internal/bundle"
"gitea.maximumdirect.net/eric/distributor/internal/config"
)
const SchemaVersion = 1
const (
SchemaVersion = 2
legacySchemaVersion = 1
StateModeSingleOwner = "single_owner"
)
type DistributorState struct {
SchemaVersion int
@@ -18,11 +23,23 @@ type DistributorState struct {
PipelineID string
DestinationID string
PublishedAt time.Time
CreatedAt time.Time
UpdatedAt time.Time
State StatePolicy
Reconciliation ReconciliationPolicy
Source SourceState
Links *LinkState
Outputs []OutputFile
}
type StatePolicy struct {
Mode string
}
type ReconciliationPolicy struct {
Mode string
}
type SourceState struct {
Manifest bundle.Manifest
}
@@ -39,6 +56,8 @@ type OutputFile struct {
URL string
SHA256 string
Size int64
CreatedAt time.Time
UpdatedAt time.Time
}
type rawDistributorState struct {
@@ -47,11 +66,23 @@ type rawDistributorState struct {
PipelineID *string `json:"pipeline_id"`
DestinationID *string `json:"destination_id"`
PublishedAt *string `json:"published_at"`
CreatedAt *string `json:"created_at"`
UpdatedAt *string `json:"updated_at"`
State *rawStatePolicy `json:"state"`
Reconciliation *rawReconciliationPolicy `json:"reconciliation"`
Source *rawSourceState `json:"source"`
Links *rawLinkState `json:"links"`
Outputs []rawOutputFile `json:"outputs"`
}
type rawStatePolicy struct {
Mode string `json:"mode"`
}
type rawReconciliationPolicy struct {
Mode string `json:"mode"`
}
type rawSourceState struct {
Manifest json.RawMessage `json:"manifest"`
}
@@ -68,6 +99,8 @@ type rawOutputFile struct {
URL string `json:"url"`
SHA256 *string `json:"sha256"`
Size *int64 `json:"size"`
CreatedAt *string `json:"created_at"`
UpdatedAt *string `json:"updated_at"`
}
func Parse(data []byte) (DistributorState, error) {
@@ -96,9 +129,10 @@ func parseRaw(raw rawDistributorState) (DistributorState, error) {
return DistributorState{}, fmt.Errorf("state schema_version is required")
}
state.SchemaVersion = *raw.SchemaVersion
if state.SchemaVersion != SchemaVersion {
return DistributorState{}, fmt.Errorf("state schema_version must be %d", SchemaVersion)
if state.SchemaVersion != SchemaVersion && state.SchemaVersion != legacySchemaVersion {
return DistributorState{}, fmt.Errorf("state schema_version must be %d or %d", legacySchemaVersion, SchemaVersion)
}
legacy := state.SchemaVersion == legacySchemaVersion
state.DistributorVersion = raw.DistributorVersion
if raw.PipelineID == nil || *raw.PipelineID == "" {
return DistributorState{}, fmt.Errorf("state pipeline_id is required")
@@ -116,6 +150,32 @@ func parseRaw(raw rawDistributorState) (DistributorState, error) {
return DistributorState{}, fmt.Errorf("state published_at must be RFC3339: %w", err)
}
state.PublishedAt = publishedAt.UTC()
if legacy {
state.SchemaVersion = SchemaVersion
state.CreatedAt = state.PublishedAt
state.UpdatedAt = state.PublishedAt
state.State.Mode = StateModeSingleOwner
state.Reconciliation.Mode = config.ReconciliationModeReplace
} else {
createdAt, err := parseRequiredTime("state created_at", raw.CreatedAt)
if err != nil {
return DistributorState{}, err
}
updatedAt, err := parseRequiredTime("state updated_at", raw.UpdatedAt)
if err != nil {
return DistributorState{}, err
}
state.CreatedAt = createdAt
state.UpdatedAt = updatedAt
if raw.State == nil || raw.State.Mode == "" {
return DistributorState{}, fmt.Errorf("state state.mode is required")
}
state.State.Mode = raw.State.Mode
if raw.Reconciliation == nil || raw.Reconciliation.Mode == "" {
return DistributorState{}, fmt.Errorf("state reconciliation.mode is required")
}
state.Reconciliation.Mode = raw.Reconciliation.Mode
}
if raw.Source == nil || len(raw.Source.Manifest) == 0 {
return DistributorState{}, fmt.Errorf("state source.manifest is required")
}
@@ -130,7 +190,7 @@ func parseRaw(raw rawDistributorState) (DistributorState, error) {
if raw.Outputs == nil {
return DistributorState{}, fmt.Errorf("state outputs is required")
}
outputs, err := parseOutputs(raw.Outputs)
outputs, err := parseOutputs(raw.Outputs, legacy, state.PublishedAt)
if err != nil {
return DistributorState{}, err
}
@@ -138,11 +198,22 @@ func parseRaw(raw rawDistributorState) (DistributorState, error) {
return state, nil
}
func parseOutputs(rawOutputs []rawOutputFile) ([]OutputFile, error) {
func parseRequiredTime(context string, raw *string) (time.Time, error) {
if raw == nil || *raw == "" {
return time.Time{}, fmt.Errorf("%s is required", context)
}
parsed, err := time.Parse(time.RFC3339, *raw)
if err != nil {
return time.Time{}, fmt.Errorf("%s must be RFC3339: %w", context, err)
}
return parsed.UTC(), nil
}
func parseOutputs(rawOutputs []rawOutputFile, legacy bool, publishedAt time.Time) ([]OutputFile, error) {
outputs := make([]OutputFile, 0, len(rawOutputs))
seen := make(map[string]struct{}, len(rawOutputs))
for index, raw := range rawOutputs {
output, err := parseOutput(index, raw)
output, err := parseOutput(index, raw, legacy, publishedAt)
if err != nil {
return nil, err
}
@@ -155,7 +226,7 @@ func parseOutputs(rawOutputs []rawOutputFile) ([]OutputFile, error) {
return outputs, nil
}
func parseOutput(index int, raw rawOutputFile) (OutputFile, error) {
func parseOutput(index int, raw rawOutputFile, legacy bool, publishedAt time.Time) (OutputFile, error) {
if raw.Path == nil || *raw.Path == "" {
return OutputFile{}, fmt.Errorf("state outputs[%d].path is required", index)
}
@@ -171,6 +242,19 @@ func parseOutput(index int, raw rawOutputFile) (OutputFile, error) {
if raw.Size == nil {
return OutputFile{}, fmt.Errorf("state outputs[%d].size is required", index)
}
createdAt := publishedAt
updatedAt := publishedAt
if !legacy {
var err error
createdAt, err = parseRequiredTime(fmt.Sprintf("state outputs[%d].created_at", index), raw.CreatedAt)
if err != nil {
return OutputFile{}, err
}
updatedAt, err = parseRequiredTime(fmt.Sprintf("state outputs[%d].updated_at", index), raw.UpdatedAt)
if err != nil {
return OutputFile{}, err
}
}
return OutputFile{
Path: *raw.Path,
Kind: *raw.Kind,
@@ -179,6 +263,8 @@ func parseOutput(index int, raw rawOutputFile) (OutputFile, error) {
URL: raw.URL,
SHA256: *raw.SHA256,
Size: *raw.Size,
CreatedAt: createdAt,
UpdatedAt: updatedAt,
}, nil
}
@@ -186,6 +272,22 @@ func (s DistributorState) PublishedAtString() string {
return s.PublishedAt.UTC().Format(time.RFC3339)
}
func (s DistributorState) CreatedAtString() string {
return s.CreatedAt.UTC().Format(time.RFC3339)
}
func (s DistributorState) UpdatedAtString() string {
return s.UpdatedAt.UTC().Format(time.RFC3339)
}
func (o OutputFile) CreatedAtString() string {
return o.CreatedAt.UTC().Format(time.RFC3339)
}
func (o OutputFile) UpdatedAtString() string {
return o.UpdatedAt.UTC().Format(time.RFC3339)
}
func (s DistributorState) MarshalJSON() ([]byte, error) {
type sourceJSON struct {
Manifest bundle.Manifest `json:"manifest"`
@@ -196,6 +298,10 @@ func (s DistributorState) MarshalJSON() ([]byte, error) {
PipelineID string `json:"pipeline_id"`
DestinationID string `json:"destination_id"`
PublishedAt string `json:"published_at"`
CreatedAt string `json:"created_at"`
UpdatedAt string `json:"updated_at"`
State StatePolicy `json:"state"`
Reconciliation ReconciliationPolicy `json:"reconciliation"`
Source sourceJSON `json:"source"`
Links *LinkState `json:"links,omitempty"`
Outputs []OutputFile `json:"outputs"`
@@ -206,12 +312,30 @@ func (s DistributorState) MarshalJSON() ([]byte, error) {
PipelineID: s.PipelineID,
DestinationID: s.DestinationID,
PublishedAt: s.PublishedAtString(),
CreatedAt: s.CreatedAtString(),
UpdatedAt: s.UpdatedAtString(),
State: s.State,
Reconciliation: s.Reconciliation,
Source: sourceJSON{Manifest: s.Source.Manifest},
Links: s.Links,
Outputs: s.Outputs,
})
}
func (p StatePolicy) MarshalJSON() ([]byte, error) {
type policyJSON struct {
Mode string `json:"mode"`
}
return json.Marshal(policyJSON{Mode: p.Mode})
}
func (p ReconciliationPolicy) MarshalJSON() ([]byte, error) {
type policyJSON struct {
Mode string `json:"mode"`
}
return json.Marshal(policyJSON{Mode: p.Mode})
}
func (l LinkState) MarshalJSON() ([]byte, error) {
type linkJSON struct {
PrimaryURL string `json:"primary_url,omitempty"`
@@ -228,6 +352,8 @@ func (o OutputFile) MarshalJSON() ([]byte, error) {
URL string `json:"url,omitempty"`
SHA256 string `json:"sha256"`
Size int64 `json:"size"`
CreatedAt string `json:"created_at"`
UpdatedAt string `json:"updated_at"`
}
return json.Marshal(outputJSON{
Path: o.Path,
@@ -237,5 +363,7 @@ func (o OutputFile) MarshalJSON() ([]byte, error) {
URL: o.URL,
SHA256: o.SHA256,
Size: o.Size,
CreatedAt: o.CreatedAt.UTC().Format(time.RFC3339),
UpdatedAt: o.UpdatedAt.UTC().Format(time.RFC3339),
})
}

139
internal/state/distributor_test.go
View File

@@ -8,6 +8,7 @@ import (
"time"
"gitea.maximumdirect.net/eric/distributor/internal/bundle"
"gitea.maximumdirect.net/eric/distributor/internal/config"
)
func TestParseValidState(t *testing.T) {
@@ -24,9 +25,24 @@ func TestParseValidState(t *testing.T) {
if got, want := state.PublishedAtString(), "2026-05-30T11:12:00Z"; got != want {
t.Fatalf("PublishedAtString() = %q, want %q", got, want)
}
if got, want := state.CreatedAtString(), "2026-05-30T11:12:00Z"; got != want {
t.Fatalf("CreatedAtString() = %q, want %q", got, want)
}
if got, want := state.UpdatedAtString(), "2026-05-30T11:12:00Z"; got != want {
t.Fatalf("UpdatedAtString() = %q, want %q", got, want)
}
if got, want := state.State.Mode, StateModeSingleOwner; got != want {
t.Fatalf("state mode = %q, want %q", got, want)
}
if got, want := state.Reconciliation.Mode, config.ReconciliationModeReplace; got != want {
t.Fatalf("reconciliation mode = %q, want %q", got, want)
}
if got, want := len(state.Outputs), 1; got != want {
t.Fatalf("output count = %d, want %d", got, want)
}
if got, want := state.Outputs[0].CreatedAtString(), "2026-05-30T11:12:00Z"; got != want {
t.Fatalf("output CreatedAtString() = %q, want %q", got, want)
}
}
func TestParseValidStateWithLinks(t *testing.T) {
@@ -62,6 +78,10 @@ func TestParseRejectsMissingFields(t *testing.T) {
"pipeline_id": `"pipeline_id"`,
"destination_id": `"destination_id"`,
"published_at": `"published_at"`,
"created_at": `"created_at"`,
"updated_at": `"updated_at"`,
"state": `"state"`,
"reconciliation": `"reconciliation"`,
"source": `"source"`,
"outputs": `"outputs"`,
}
@@ -75,9 +95,31 @@ func TestParseRejectsMissingFields(t *testing.T) {
}
func TestParseRejectsInvalidSchemaVersion(t *testing.T) {
body := strings.Replace(validStateJSON(t), `"schema_version": 1`, `"schema_version": 2`, 1)
body := strings.Replace(validStateJSON(t), `"schema_version": 2`, `"schema_version": 3`, 1)
_, err := Parse([]byte(body))
assertStateErrorContains(t, err, "schema_version must be 1")
assertStateErrorContains(t, err, "schema_version must be 1 or 2")
}
func TestParseLegacyStateInfersSingleOwnerDefaults(t *testing.T) {
state, err := Parse([]byte(legacyStateJSON(t)))
if err != nil {
t.Fatalf("Parse() error = %v", err)
}
if got, want := state.SchemaVersion, SchemaVersion; got != want {
t.Fatalf("schema version = %d, want normalized %d", got, want)
}
if got, want := state.CreatedAtString(), "2026-05-30T11:12:00Z"; got != want {
t.Fatalf("created_at = %q, want %q", got, want)
}
if got, want := state.State.Mode, StateModeSingleOwner; got != want {
t.Fatalf("state mode = %q, want %q", got, want)
}
if got, want := state.Reconciliation.Mode, config.ReconciliationModeReplace; got != want {
t.Fatalf("reconciliation mode = %q, want %q", got, want)
}
if got, want := state.Outputs[0].UpdatedAtString(), "2026-05-30T11:12:00Z"; got != want {
t.Fatalf("output updated_at = %q, want %q", got, want)
}
}
func TestParseRejectsInvalidEmbeddedManifest(t *testing.T) {
@@ -161,6 +203,12 @@ func TestParseRejectsInvalidOutputMetadata(t *testing.T) {
"negative size": func(s *DistributorState) {
s.Outputs[0].Size = -1
},
"missing output created_at": func(s *DistributorState) {
s.Outputs[0].CreatedAt = time.Time{}
},
"missing output updated_at": func(s *DistributorState) {
s.Outputs[0].UpdatedAt = time.Time{}
},
"invalid output url": func(s *DistributorState) {
s.Outputs[0].URL = "file:///tmp/report.md"
},
@@ -202,13 +250,24 @@ func TestParseRejectsMalformedPublishedTimestamp(t *testing.T) {
assertStateErrorContains(t, err, "published_at must be RFC3339")
}
func TestParseRejectsMalformedCreatedTimestamp(t *testing.T) {
body := strings.Replace(validStateJSON(t), `"created_at": "2026-05-30T11:12:00Z"`, `"created_at": "May 30"`, 1)
_, err := Parse([]byte(body))
assertStateErrorContains(t, err, "created_at must be RFC3339")
}
func TestMarshalNormalizesPublishedAtUTC(t *testing.T) {
source := validManifest(t)
publishedAt := time.Date(2026, 5, 30, 13, 12, 0, 0, time.FixedZone("offset", 2*60*60))
state := DistributorState{
SchemaVersion: SchemaVersion,
PipelineID: "reports",
DestinationID: "archive",
PublishedAt: time.Date(2026, 5, 30, 13, 12, 0, 0, time.FixedZone("offset", 2*60*60)),
PublishedAt: publishedAt,
CreatedAt: publishedAt,
UpdatedAt: publishedAt,
State: StatePolicy{Mode: StateModeSingleOwner},
Reconciliation: ReconciliationPolicy{Mode: config.ReconciliationModeReplace},
Source: SourceState{Manifest: source},
Outputs: []OutputFile{{
Path: "report.md",
@@ -216,6 +275,8 @@ func TestMarshalNormalizesPublishedAtUTC(t *testing.T) {
SourcePath: "report.md",
SHA256: source.Files[0].SHA256,
Size: source.Files[0].Size,
CreatedAt: publishedAt,
UpdatedAt: publishedAt,
}},
}
data, err := json.Marshal(state)
@@ -229,11 +290,16 @@ func TestMarshalNormalizesPublishedAtUTC(t *testing.T) {
func TestMarshalIncludesLinksWhenPresent(t *testing.T) {
source := validManifest(t)
publishedAt := time.Date(2026, 5, 30, 11, 12, 0, 0, time.UTC)
state := DistributorState{
SchemaVersion: SchemaVersion,
PipelineID: "reports",
DestinationID: "archive",
PublishedAt: time.Date(2026, 5, 30, 11, 12, 0, 0, time.UTC),
PublishedAt: publishedAt,
CreatedAt: publishedAt,
UpdatedAt: publishedAt,
State: StatePolicy{Mode: StateModeSingleOwner},
Reconciliation: ReconciliationPolicy{Mode: config.ReconciliationModeReplace},
Source: SourceState{Manifest: source},
Links: &LinkState{PrimaryURL: "https://reports.example.com/archive/report.md"},
Outputs: []OutputFile{{
@@ -243,6 +309,8 @@ func TestMarshalIncludesLinksWhenPresent(t *testing.T) {
URL: "https://reports.example.com/archive/report.md",
SHA256: source.Files[0].SHA256,
Size: source.Files[0].Size,
CreatedAt: publishedAt,
UpdatedAt: publishedAt,
}},
}
data, err := json.Marshal(state)
@@ -257,6 +325,42 @@ func TestMarshalIncludesLinksWhenPresent(t *testing.T) {
}
}
func TestOutputHelpers(t *testing.T) {
createdAt := time.Date(2026, 5, 30, 11, 12, 0, 0, time.UTC)
updatedAt := createdAt.Add(time.Hour)
retained := []OutputFile{{
Path: "old.md",
Kind: OutputKindSource,
CreatedAt: createdAt,
UpdatedAt: createdAt,
}, {
Path: "report.md",
Kind: OutputKindSource,
CreatedAt: createdAt,
UpdatedAt: createdAt,
}}
projected := ProjectOutputs([]OutputProjection{{
Path: "report.md",
Kind: OutputKindSource,
}, {
Path: "new.md",
Kind: OutputKindSource,
}}, retained, updatedAt)
if got, ok := FindOutputByPath(projected, "report.md"); !ok || !got.CreatedAt.Equal(createdAt) || !got.UpdatedAt.Equal(updatedAt) {
t.Fatalf("projected report.md = %#v, want preserved created_at and updated updated_at", got)
}
merged, err := MergeOutputFiles(retained, projected)
if err != nil {
t.Fatalf("MergeOutputFiles() error = %v", err)
}
if got, want := len(merged), 3; got != want {
t.Fatalf("merged count = %d, want %d", got, want)
}
if got, want := ManagedOutputPaths(DistributorState{Outputs: merged}), []string{"old.md", "report.md", "new.md"}; strings.Join(got, ",") != strings.Join(want, ",") {
t.Fatalf("managed paths = %#v, want %#v", got, want)
}
}
func validStateJSON(t *testing.T) string {
t.Helper()
return validStateWithManifestJSON(t, manifestJSON(t))
@@ -265,11 +369,15 @@ func validStateJSON(t *testing.T) string {
func validStateWithManifestJSON(t *testing.T, manifest string) string {
t.Helper()
return `{
"schema_version": 1,
"schema_version": 2,
"distributor_version": "dev",
"pipeline_id": "reports",
"destination_id": "archive",
"published_at": "2026-05-30T11:12:00Z",
"created_at": "2026-05-30T11:12:00Z",
"updated_at": "2026-05-30T11:12:00Z",
"state": {"mode": "single_owner"},
"reconciliation": {"mode": "replace"},
"source": {
"manifest": ` + manifest + `
},
@@ -279,12 +387,31 @@ func validStateWithManifestJSON(t *testing.T, manifest string) string {
"kind": "source",
"source_path": "report.md",
"sha256": "sha256:3640fd37140ee4d2e0e93e78834f232ea67a50e7bc6279203690cc7de1975fa6",
"size": 16
"size": 16,
"created_at": "2026-05-30T11:12:00Z",
"updated_at": "2026-05-30T11:12:00Z"
}
]
}`
}
func legacyStateJSON(t *testing.T) string {
t.Helper()
return strings.Replace(strings.Replace(strings.Replace(strings.Replace(strings.Replace(strings.Replace(validStateJSON(t),
`"schema_version": 2`, `"schema_version": 1`, 1),
` "created_at": "2026-05-30T11:12:00Z",
`, "", 1),
` "updated_at": "2026-05-30T11:12:00Z",
`, "", 1),
` "state": {"mode": "single_owner"},
`, "", 1),
` "reconciliation": {"mode": "replace"},
`, "", 1),
`,
"created_at": "2026-05-30T11:12:00Z",
"updated_at": "2026-05-30T11:12:00Z"`, "", 1)
}
func manifestJSON(t *testing.T) string {
t.Helper()
data, err := os.ReadFile("../bundle/testdata/valid_bundle/manifest.json")

82
internal/state/outputs.go Normal file
View File

@@ -0,0 +1,82 @@
package state
import (
"fmt"
"time"
)
type OutputProjection struct {
Path string
Kind string
SourcePath string
Transform string
URL string
SHA256 string
Size int64
}
func FindOutputByPath(outputs []OutputFile, path string) (OutputFile, bool) {
for _, output := range outputs {
if output.Path == path {
return output, true
}
}
return OutputFile{}, false
}
func MergeOutputFiles(retained, planned []OutputFile) ([]OutputFile, error) {
outputs := make([]OutputFile, 0, len(retained)+len(planned))
indexByPath := make(map[string]int, len(retained)+len(planned))
for _, output := range retained {
if _, exists := indexByPath[output.Path]; exists {
return nil, fmt.Errorf("state output path %q is duplicated", output.Path)
}
indexByPath[output.Path] = len(outputs)
outputs = append(outputs, output)
}
seenPlanned := make(map[string]struct{}, len(planned))
for _, output := range planned {
if _, exists := seenPlanned[output.Path]; exists {
return nil, fmt.Errorf("state output path %q is duplicated", output.Path)
}
seenPlanned[output.Path] = struct{}{}
if index, exists := indexByPath[output.Path]; exists {
outputs[index] = output
continue
}
indexByPath[output.Path] = len(outputs)
outputs = append(outputs, output)
}
return outputs, nil
}
func ManagedOutputPaths(s DistributorState) []string {
paths := make([]string, 0, len(s.Outputs))
for _, output := range s.Outputs {
paths = append(paths, output.Path)
}
return paths
}
func ProjectOutputs(outputs []OutputProjection, existing []OutputFile, now time.Time) []OutputFile {
now = now.UTC()
files := make([]OutputFile, 0, len(outputs))
for _, output := range outputs {
createdAt := now
if existingOutput, ok := FindOutputByPath(existing, output.Path); ok {
createdAt = existingOutput.CreatedAt
}
files = append(files, OutputFile{
Path: output.Path,
Kind: output.Kind,
SourcePath: output.SourcePath,
Transform: output.Transform,
URL: output.URL,
SHA256: output.SHA256,
Size: output.Size,
CreatedAt: createdAt,
UpdatedAt: now,
})
}
return files
}

19
internal/state/validate.go
View File

@@ -4,6 +4,7 @@ import (
"fmt"
"gitea.maximumdirect.net/eric/distributor/internal/bundle"
"gitea.maximumdirect.net/eric/distributor/internal/config"
"gitea.maximumdirect.net/eric/distributor/internal/link"
"gitea.maximumdirect.net/eric/distributor/internal/storage"
)
@@ -26,6 +27,18 @@ func Validate(s DistributorState) error {
if s.PublishedAt.IsZero() {
return fmt.Errorf("state published_at is required")
}
if s.CreatedAt.IsZero() {
return fmt.Errorf("state created_at is required")
}
if s.UpdatedAt.IsZero() {
return fmt.Errorf("state updated_at is required")
}
if s.State.Mode != StateModeSingleOwner {
return fmt.Errorf("state state.mode must be %s", StateModeSingleOwner)
}
if s.Reconciliation.Mode != config.ReconciliationModeReplace && s.Reconciliation.Mode != config.ReconciliationModeMerge {
return fmt.Errorf("state reconciliation.mode must be %s or %s", config.ReconciliationModeReplace, config.ReconciliationModeMerge)
}
if err := validateEmbeddedManifest(s.Source.Manifest); err != nil {
return fmt.Errorf("state source.manifest: %w", err)
}
@@ -80,5 +93,11 @@ func validateOutput(index int, output OutputFile) error {
if output.Size < 0 {
return fmt.Errorf("state outputs[%d].size must be non-negative", index)
}
if output.CreatedAt.IsZero() {
return fmt.Errorf("state outputs[%d].created_at is required", index)
}
if output.UpdatedAt.IsZero() {
return fmt.Errorf("state outputs[%d].updated_at is required", index)
}
return nil
}

14
internal/testutil/fixtures.go
View File

@@ -11,6 +11,7 @@ import (
"time"
"gitea.maximumdirect.net/eric/distributor/internal/bundle"
"gitea.maximumdirect.net/eric/distributor/internal/config"
"gitea.maximumdirect.net/eric/distributor/internal/state"
"gitea.maximumdirect.net/eric/distributor/internal/storage"
"gitea.maximumdirect.net/eric/distributor/internal/storage/fake"
@@ -329,14 +330,19 @@ func WriteDestinationState(t testing.TB, root, relative string, manifest bundle.
}
func DestinationState(manifest bundle.Manifest, opts DestinationStateOptions) state.DistributorState {
publishedAt := defaultPublishedAt(opts.PublishedAt)
return state.DistributorState{
SchemaVersion: state.SchemaVersion,
DistributorVersion: opts.DistributorVersion,
PipelineID: defaultString(opts.PipelineID, "reports"),
DestinationID: defaultString(opts.DestinationID, "archive"),
PublishedAt: defaultPublishedAt(opts.PublishedAt),
PublishedAt: publishedAt,
CreatedAt: publishedAt,
UpdatedAt: publishedAt,
State: state.StatePolicy{Mode: state.StateModeSingleOwner},
Reconciliation: state.ReconciliationPolicy{Mode: config.ReconciliationModeReplace},
Source: state.SourceState{Manifest: manifest},
Outputs: sourceOutputs(manifest),
Outputs: sourceOutputs(manifest, publishedAt),
}
}
@@ -386,7 +392,7 @@ func sourceFiles(opts BundleOptions) []SourceFile {
return files
}
func sourceOutputs(manifest bundle.Manifest) []state.OutputFile {
func sourceOutputs(manifest bundle.Manifest, publishedAt time.Time) []state.OutputFile {
outputs := make([]state.OutputFile, 0, len(manifest.Files))
for _, file := range manifest.Files {
outputs = append(outputs, state.OutputFile{
@@ -395,6 +401,8 @@ func sourceOutputs(manifest bundle.Manifest) []state.OutputFile {
SourcePath: file.Path,
SHA256: file.SHA256,
Size: file.Size,
CreatedAt: publishedAt,
UpdatedAt: publishedAt,
})
}
return outputs