diff --git a/Dockerfile.api b/Dockerfile.api index acd081f..213bc5c 100644 --- a/Dockerfile.api +++ b/Dockerfile.api @@ -1,9 +1,12 @@ FROM golang:1.22-alpine AS build WORKDIR /src +ARG BUILD_REVISION=devel +ARG BUILD_TIME=unknown +ARG BUILD_DIRTY=unknown COPY go.mod ./ RUN go mod download COPY . ./ -RUN go build -trimpath -ldflags='-s -w' -o /out/orchestra ./cmd/orchestra +RUN go build -trimpath -ldflags="-s -w -X orchestra/internal/buildinfo.Revision=${BUILD_REVISION} -X orchestra/internal/buildinfo.Time=${BUILD_TIME} -X orchestra/internal/buildinfo.Dirty=${BUILD_DIRTY}" -o /out/orchestra ./cmd/orchestra FROM alpine:3.21 RUN adduser -D -u 10001 orchestra diff --git a/cmd/orchestra-migrate/main.go b/cmd/orchestra-migrate/main.go new file mode 100644 index 0000000..2067bbb --- /dev/null +++ b/cmd/orchestra-migrate/main.go @@ -0,0 +1,29 @@ +// orchestra-migrate contains explicit, one-shot durable-store migrations. +// It deliberately never starts the coordinator: run it only while all +// coordinator processes for the target data directory are stopped. +package main + +import ( + "flag" + "fmt" + "log" + "orchestra/internal/store" +) + +func main() { + dir := flag.String("data", "", "Orchestra data directory") + confirm := flag.Bool("confirm", false, "confirm that every coordinator using -data is stopped") + flag.Parse() + if *dir == "" || !*confirm { + log.Fatal("usage: orchestra-migrate -data DIR -confirm (with all coordinators stopped)") + } + changed, err := store.NormalizeLegacyEventSequence(*dir) + if err != nil { + log.Fatal(err) + } + if changed { + fmt.Println("normalized legacy event sequence; preserved events.jsonl.legacy-* backup") + return + } + fmt.Println("event sequence already canonical; no migration needed") +} diff --git a/deploy/DEPLOYMENT.md b/deploy/DEPLOYMENT.md index d3a4d2d..8d72237 100644 --- a/deploy/DEPLOYMENT.md +++ b/deploy/DEPLOYMENT.md @@ -41,3 +41,28 @@ Verify the coordinator at `GET /v1/admin/diagnostics` with the normal admin credential: its `build` object is the coordinator provenance. `GET /v1/federation/workers` shows every worker's `build`, supported projects, and worker-local health without SSH. + +For the Docker coordinator deployment, provide the same provenance as build +arguments (the Dockerfile intentionally cannot read `.git` from its build +context): + +```sh +revision=$(git rev-parse HEAD) +build_time=$(date -u +%Y-%m-%dT%H:%M:%SZ) +dirty=false; test -z "$(git status --porcelain)" || dirty=true +docker compose build \ + --build-arg BUILD_REVISION="$revision" \ + --build-arg BUILD_TIME="$build_time" \ + --build-arg BUILD_DIRTY="$dirty" \ + orchestra-api +docker compose up -d --no-deps orchestra-api +``` + +If a pre-v2 event log has the historical repeated-`seq=1` prefix, the current +coordinator intentionally refuses to replay it. Stop every coordinator using +the data directory and run the explicit, backup-preserving migration before +deploying the current image: + +```sh +orchestra-migrate -data /var/lib/orchestra/data -confirm +``` diff --git a/internal/store/migration_test.go b/internal/store/migration_test.go new file mode 100644 index 0000000..7f5339a --- /dev/null +++ b/internal/store/migration_test.go @@ -0,0 +1,58 @@ +package store + +import ( + "encoding/json" + "os" + "path/filepath" + "testing" + "time" + + "orchestra/internal/domain" +) + +func TestNormalizeLegacyEventSequence(t *testing.T) { + dir := t.TempDir() + events := []domain.Event{ + {SchemaVersion: 2, Seq: 1, ID: "created", Type: "TaskCreated", TaskID: "t", Version: 1, At: time.Now().UTC(), Payload: json.RawMessage(`{"source":"test","external_id":"1","project":"p"}`), Surface: "system"}, + {SchemaVersion: 2, Seq: 1, ID: "leased", Type: "TaskLeased", TaskID: "t", Version: 2, At: time.Now().UTC(), Payload: json.RawMessage(`{"harness_id":"h","ttl":60,"until_ns":2000000000000000000,"expected_version":1}`), Surface: "system"}, + {SchemaVersion: 2, Seq: 2, ID: "recovered-lease", Type: "TaskLeased", TaskID: "t", Version: 2, At: time.Now().UTC(), Payload: json.RawMessage(`{"harness_id":"h","ttl":60,"until_ns":2000000000000000000,"expected_version":1}`), Surface: "system"}, + } + var raw []byte + for _, e := range events { + b, err := json.Marshal(e) + if err != nil { + t.Fatal(err) + } + raw = append(raw, append(b, '\n')...) + } + if err := os.WriteFile(filepath.Join(dir, "events.jsonl"), raw, 0644); err != nil { + t.Fatal(err) + } + changed, err := NormalizeLegacyEventSequence(dir) + if err != nil || !changed { + t.Fatalf("NormalizeLegacyEventSequence = %v, %v", changed, err) + } + if backups, err := filepath.Glob(filepath.Join(dir, "events.jsonl.legacy-*")); err != nil || len(backups) != 1 { + t.Fatalf("backups = %v, %v", backups, err) + } + s, err := Open(dir) + if err != nil { + t.Fatal(err) + } + if got := len(s.Events(0)); got != 2 { + t.Fatalf("events = %d, want 2", got) + } + if got := s.Events(0)[1].ID; got != "recovered-lease" { + t.Fatalf("second event = %q, want recovered suffix", got) + } +} + +func TestNormalizeLegacyEventSequenceRejectsOtherCorruption(t *testing.T) { + dir := t.TempDir() + if err := os.WriteFile(filepath.Join(dir, "events.jsonl"), []byte(`{"seq":2}`+"\n"), 0644); err != nil { + t.Fatal(err) + } + if _, err := NormalizeLegacyEventSequence(dir); err == nil { + t.Fatal("NormalizeLegacyEventSequence accepted non-legacy corruption") + } +} diff --git a/internal/store/store.go b/internal/store/store.go index 7347fcb..9373d1a 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -2,6 +2,7 @@ package store import ( "bufio" + "bytes" "encoding/json" "errors" "fmt" @@ -94,6 +95,125 @@ func Open(dir string) (*Store, error) { } return s, sc.Err() } + +// NormalizeLegacyEventSequence repairs one explicitly recognized pre-v2 log +// shape: a prefix of two or more records all numbered seq=1, followed by a +// contiguous suffix numbered 2, 3, ... that restarts the first task at +// version 2. Early Orchestra releases emitted precisely that shape: records +// after the first seq=1 were never included in the compatibility snapshot, +// then were replayed again after restart. The migration retains the initial +// TaskCreated plus the contiguous suffix, discarding only the provably +// abandoned duplicate prefix. It is not a general corruption repair tool: +// any other gap or duplicate is rejected so Open's fail-closed recovery +// guarantee remains intact. +// +// The caller must stop every coordinator using dir first. The original log is +// durably copied to events.jsonl.legacy- before an fsync+rename +// replacement is installed. The return value reports whether a migration was +// needed. +func NormalizeLegacyEventSequence(dir string) (bool, error) { + path := filepath.Join(dir, "events.jsonl") + raw, err := os.ReadFile(path) + if err != nil { + return false, err + } + lines := bytes.Split(bytes.TrimSpace(raw), []byte{'\n'}) + if len(lines) == 0 || (len(lines) == 1 && len(lines[0]) == 0) { + return false, nil + } + events := make([]domain.Event, len(lines)) + for i, line := range lines { + if err := json.Unmarshal(line, &events[i]); err != nil { + return false, fmt.Errorf("event %d: %w", i+1, err) + } + } + canonical := true + for i, e := range events { + if e.Seq != uint64(i+1) { + canonical = false + break + } + } + if canonical { + return false, nil + } + prefix := 0 + for prefix < len(events) && events[prefix].Seq == 1 { + prefix++ + } + if prefix < 2 { + return false, fmt.Errorf("refusing non-legacy event sequence") + } + if events[0].Type != "TaskCreated" || events[0].Version != 1 { + return false, fmt.Errorf("refusing legacy sequence without an initial task creation") + } + for i := 1; i < prefix; i++ { + if events[i].TaskID != events[0].TaskID || events[i].Version < 2 { + return false, fmt.Errorf("refusing non-legacy duplicate prefix at record %d", i+1) + } + } + if prefix == len(events) || events[prefix].Type != "TaskLeased" || events[prefix].TaskID != events[0].TaskID || events[prefix].Version != 2 { + return false, fmt.Errorf("refusing legacy sequence without a task-version-2 restart") + } + for i := prefix; i < len(events); i++ { + want := uint64(i - prefix + 2) + if events[i].Seq != want { + return false, fmt.Errorf("refusing non-legacy event sequence at record %d: got %d, want %d", i+1, events[i].Seq, want) + } + } + backup := fmt.Sprintf("%s.legacy-%d", path, time.Now().UTC().UnixNano()) + backupFile, err := os.OpenFile(backup, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0644) + if err != nil { + return false, err + } + if _, err = backupFile.Write(raw); err == nil { + err = backupFile.Sync() + } + if closeErr := backupFile.Close(); err == nil { + err = closeErr + } + if err != nil { + return false, err + } + tmp := path + ".sequence-migration.tmp" + out, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0644) + if err != nil { + return false, err + } + kept := append(events[:1:1], events[prefix:]...) + for i := range kept { + kept[i].Seq = uint64(i + 1) + line, marshalErr := json.Marshal(kept[i]) + if marshalErr != nil { + err = marshalErr + break + } + if _, err = out.Write(append(line, '\n')); err != nil { + break + } + } + if err == nil { + err = out.Sync() + } + if closeErr := out.Close(); err == nil { + err = closeErr + } + if err != nil { + return false, err + } + if err := os.Rename(tmp, path); err != nil { + return false, err + } + d, err := os.Open(dir) + if err != nil { + return false, err + } + defer d.Close() + if err := d.Sync(); err != nil { + return false, err + } + return true, nil +} func (s *Store) apply(e domain.Event) error { var p map[string]any if err := json.Unmarshal(e.Payload, &p); err != nil {