feat(syncjob): 新增可恢复的数据同步任务运行时

- 持久化任务、运行、事件、检查点与错误行状态
- 支持调度、暂停取消、恢复重试、并发限制和故障接管
- 增加执行 fencing、连续失败退避与错误行重放租约
This commit is contained in:
Syngnat
2026-08-08 19:06:57 +08:00
parent e40cdbc016
commit bc02a4c614
13 changed files with 6309 additions and 0 deletions

View File

@@ -0,0 +1,78 @@
package syncjob
import (
"context"
"hash/fnv"
"time"
)
const (
continuousFailureBackoffInitial = 5 * time.Second
continuousFailureBackoffMaximum = 5 * time.Minute
continuousFailureHistoryLimit = 32
)
func (m *Manager) continuousFailureNotBefore(ctx context.Context, definition JobDefinition) (int64, int, error) {
if definition.Schedule.Kind != ScheduleContinuous {
return 0, 0, nil
}
runs, err := m.store.ListRuns(ctx, definition.ID, continuousFailureHistoryLimit)
if err != nil {
return 0, 0, err
}
consecutiveFailures := 0
var latestFailure RunRecord
for _, run := range runs {
switch run.Status {
case RunStatusQueued, RunStatusRunning, RunStatusCancelling:
continue
case RunStatusFailed, RunStatusPartial, RunStatusInterrupted:
if consecutiveFailures == 0 {
latestFailure = run
}
consecutiveFailures++
default:
// A success or an operator-controlled terminal state ends the failure
// streak. The next continuous launch returns to the normal poll cadence.
goto counted
}
}
counted:
if consecutiveFailures == 0 || latestFailure.FinishedAt <= 0 {
return 0, consecutiveFailures, nil
}
backoff := continuousFailureBackoff(definition.ID, latestFailure.ID, consecutiveFailures)
return time.UnixMilli(latestFailure.FinishedAt).Add(backoff).UnixMilli(), consecutiveFailures, nil
}
func continuousFailureBackoff(jobID, latestRunID string, consecutiveFailures int) time.Duration {
if consecutiveFailures < 1 {
return 0
}
base := continuousFailureBackoffInitial
for attempt := 1; attempt < consecutiveFailures && base < continuousFailureBackoffMaximum; attempt++ {
if base > continuousFailureBackoffMaximum/2 {
base = continuousFailureBackoffMaximum
break
}
base *= 2
}
if base >= continuousFailureBackoffMaximum {
return continuousFailureBackoffMaximum
}
jitterRoom := base / 5
if remaining := continuousFailureBackoffMaximum - base; jitterRoom > remaining {
jitterRoom = remaining
}
if jitterRoom <= 0 {
return base
}
hasher := fnv.New64a()
_, _ = hasher.Write([]byte(jobID))
_, _ = hasher.Write([]byte{0})
_, _ = hasher.Write([]byte(latestRunID))
_, _ = hasher.Write([]byte{0, byte(consecutiveFailures)})
jitter := time.Duration(hasher.Sum64() % (uint64(jitterRoom) + 1))
return base + jitter
}

View File

@@ -0,0 +1,169 @@
package syncjob
import (
"context"
"encoding/json"
"errors"
"testing"
"time"
)
func TestContinuousFailureBackoffPersistsAcrossRestartAndSuccessResets(t *testing.T) {
path := t.TempDir() + "/continuous.db"
store, err := Open(path)
if err != nil {
t.Fatalf("open store: %v", err)
}
definition := putContinuousTestJob(t, store)
failedAt := time.Now().Truncate(time.Millisecond)
createTerminalHistoryRun(t, store, definition, RunStatusFailed, failedAt)
now := failedAt.Add(time.Second)
if _, err := store.db.ExecContext(context.Background(), `UPDATE data_sync_jobs SET next_run_at = ? WHERE id = ?`, now.Add(-time.Millisecond).UnixMilli(), definition.ID); err != nil {
t.Fatalf("make continuous job due: %v", err)
}
manager := newManualSchedulerManager(store, now, "first-owner")
notBefore, failures, err := manager.continuousFailureNotBefore(context.Background(), definition)
if err != nil || failures != 1 {
t.Fatalf("failure backoff = %d, failures=%d, err=%v", notBefore, failures, err)
}
delay := time.Duration(notBefore-failedAt.UnixMilli()) * time.Millisecond
if delay < 5*time.Second || delay > 6*time.Second {
t.Fatalf("first failure delay = %s, want [5s, 6s]", delay)
}
manager.runSchedulerCycle()
delayed, err := store.GetJob(context.Background(), definition.ID)
if err != nil || delayed.NextRunAt != notBefore {
t.Fatalf("delayed job = %#v, err=%v", delayed, err)
}
runs, err := store.ListRuns(context.Background(), definition.ID, 10)
if err != nil || len(runs) != 1 {
t.Fatalf("runs during backoff = %#v, err=%v", runs, err)
}
_ = store.ReleaseSchedulerLease(context.Background(), "data-sync-scheduler", "first-owner")
if err := store.Close(); err != nil {
t.Fatalf("close store: %v", err)
}
reopened, err := Open(path)
if err != nil {
t.Fatalf("reopen store: %v", err)
}
t.Cleanup(func() { _ = reopened.Close() })
restarted := newManualSchedulerManager(reopened, now, "restart-owner")
reloaded, err := reopened.GetJob(context.Background(), definition.ID)
if err != nil || reloaded.NextRunAt != notBefore {
t.Fatalf("reloaded delayed job = %#v, err=%v", reloaded, err)
}
restartedNotBefore, restartedFailures, err := restarted.continuousFailureNotBefore(context.Background(), reloaded)
if err != nil || restartedFailures != failures || restartedNotBefore != notBefore {
t.Fatalf("restart backoff = %d/%d, want %d/%d, err=%v", restartedNotBefore, restartedFailures, notBefore, failures, err)
}
restarted.runSchedulerCycle()
runs, err = reopened.ListRuns(context.Background(), definition.ID, 10)
if err != nil || len(runs) != 1 {
t.Fatalf("restart runs during backoff = %#v, err=%v", runs, err)
}
createTerminalHistoryRun(t, reopened, reloaded, RunStatusSucceeded, failedAt.Add(2*time.Second))
resetAt, resetFailures, err := restarted.continuousFailureNotBefore(context.Background(), reloaded)
if err != nil || resetAt != 0 || resetFailures != 0 {
t.Fatalf("success reset backoff = %d, failures=%d, err=%v", resetAt, resetFailures, err)
}
}
func TestContinuousFailureBackoffIsStableExponentialAndCapped(t *testing.T) {
previous := time.Duration(0)
for failures := 1; failures <= 12; failures++ {
first := continuousFailureBackoff("job", "run", failures)
second := continuousFailureBackoff("job", "run", failures)
if first != second {
t.Fatalf("failure %d jitter is not stable: %s != %s", failures, first, second)
}
if first < previous || first > 5*time.Minute {
t.Fatalf("failure %d backoff = %s, previous=%s", failures, first, previous)
}
previous = first
}
if previous != 5*time.Minute {
t.Fatalf("capped backoff = %s, want 5m", previous)
}
}
func TestPermanentExecutionErrorPausesOwningJob(t *testing.T) {
store := openTestStore(t)
definition := putTestJob(t, store, "queue")
manager := newTestManager(t, store, ExecutorFunc(func(context.Context, ExecutionRequest, RunReporter) (ExecutionOutcome, error) {
return ExecutionOutcome{}, MarkPermanentExecutionError(errors.New("unsupported source topology"))
}))
run, err := manager.StartRun(context.Background(), definition.ID)
if err != nil {
t.Fatalf("start run: %v", err)
}
waitRunStatus(t, store, run.ID, RunStatusFailed)
deadline := time.Now().Add(time.Second)
for time.Now().Before(deadline) {
paused, getErr := store.GetJob(context.Background(), definition.ID)
if getErr == nil && paused.Lifecycle == JobLifecyclePaused && !paused.Enabled {
if _, err := manager.StartRun(context.Background(), definition.ID); !errors.Is(err, ErrJobDisabled) {
t.Fatalf("start paused job error = %v, want ErrJobDisabled", err)
}
return
}
time.Sleep(time.Millisecond)
}
t.Fatal("permanent failure did not pause job")
}
func putContinuousTestJob(t *testing.T, store *Store) JobDefinition {
t.Helper()
definition := JobDefinition{
Name: "continuous orders", Lifecycle: JobLifecycleEnabled, Enabled: true, Kind: JobKindReconcile,
IncrementalMode: IncrementalCDC, Source: EndpointRef{ConnectionID: "source"}, Target: EndpointRef{ConnectionID: "target"},
Mappings: []TableMapping{{SourceTable: "orders", TargetTable: "orders", KeyColumns: []string{"id"}, Enabled: true}},
CDC: &CDCSpec{Adapter: "mongodb-change-stream", StartPosition: "checkpoint"},
Schedule: ScheduleSpec{Kind: ScheduleContinuous}, ConcurrencyPolicy: "forbid",
}
saved, err := store.PutJob(context.Background(), definition)
if err != nil {
t.Fatalf("put continuous job: %v", err)
}
return saved
}
func createTerminalHistoryRun(t *testing.T, store *Store, definition JobDefinition, status RunStatus, finishedAt time.Time) RunRecord {
t.Helper()
snapshot, err := json.Marshal(definition)
if err != nil {
t.Fatalf("marshal definition: %v", err)
}
run, err := store.CreateRun(context.Background(), RunRecord{
JobID: definition.ID, JobRevision: definition.Revision, Status: RunStatusRunning, DefinitionSnapshot: snapshot,
})
if err != nil {
t.Fatalf("create history run: %v", err)
}
run, err = store.CompleteRun(context.Background(), run.ID, status, ExecutionOutcome{}, string(status), finishedAt.UnixMilli())
if err != nil {
t.Fatalf("complete history run: %v", err)
}
if _, err := store.db.ExecContext(context.Background(), `UPDATE data_sync_runs SET created_at = ?, updated_at = ? WHERE id = ?`,
finishedAt.UnixMilli(), finishedAt.UnixMilli(), run.ID); err != nil {
t.Fatalf("order history run: %v", err)
}
return run
}
func newManualSchedulerManager(store *Store, now time.Time, owner string) *Manager {
ctx, cancel := context.WithCancelCause(context.Background())
options := normalizeManagerOptions(ManagerOptions{
SchedulerInterval: time.Hour, LeaseTTL: time.Minute, HeartbeatInterval: time.Hour,
RecoveryStaleAfter: time.Hour, RecoveryInterval: time.Hour, LeaseOwner: owner, Now: func() time.Time { return now },
})
return &Manager{
store: store, executor: ExecutorFunc(func(context.Context, ExecutionRequest, RunReporter) (ExecutionOutcome, error) {
return ExecutionOutcome{}, nil
}),
options: options, ctx: ctx, cancel: cancel, wake: make(chan struct{}, 1), active: make(map[string]activeExecution),
lastRecoveryAt: now, done: make(chan struct{}),
}
}

1159
internal/syncjob/manager.go Normal file

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,883 @@
package syncjob
import (
"context"
"encoding/json"
"errors"
"sync"
"sync/atomic"
"testing"
"time"
)
func TestManagerQueuesRunsForTheSameJobWithoutOverlap(t *testing.T) {
store := openTestStore(t)
definition := putTestJob(t, store, "queue")
started := make(chan string, 2)
release := make(chan struct{}, 2)
var active atomic.Int32
var maxActive atomic.Int32
executor := ExecutorFunc(func(ctx context.Context, request ExecutionRequest, _ RunReporter) (ExecutionOutcome, error) {
current := active.Add(1)
defer active.Add(-1)
for {
maximum := maxActive.Load()
if current <= maximum || maxActive.CompareAndSwap(maximum, current) {
break
}
}
started <- request.Run.ID
select {
case <-ctx.Done():
return ExecutionOutcome{}, context.Cause(ctx)
case <-release:
return ExecutionOutcome{RowsInserted: 1}, nil
}
})
manager, err := NewManager(context.Background(), store, executor, ManagerOptions{
SchedulerInterval: time.Hour,
HeartbeatInterval: time.Hour,
})
if err != nil {
t.Fatalf("new manager: %v", err)
}
t.Cleanup(func() {
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
if err := manager.Shutdown(ctx); err != nil {
t.Errorf("shutdown manager: %v", err)
}
})
first, err := manager.StartRun(context.Background(), definition.ID)
if err != nil {
t.Fatalf("start first run: %v", err)
}
if got := receiveString(t, started); got != first.ID {
t.Fatalf("first executed run = %q, want %q", got, first.ID)
}
second, err := manager.StartRun(context.Background(), definition.ID)
if err != nil {
t.Fatalf("queue second run: %v", err)
}
assertRunStatus(t, store, second.ID, RunStatusQueued)
select {
case got := <-started:
t.Fatalf("second run %q overlapped first run", got)
case <-time.After(75 * time.Millisecond):
}
release <- struct{}{}
waitRunStatus(t, store, first.ID, RunStatusSucceeded)
if got := receiveString(t, started); got != second.ID {
t.Fatalf("second executed run = %q, want %q", got, second.ID)
}
release <- struct{}{}
waitRunStatus(t, store, second.ID, RunStatusSucceeded)
if got := maxActive.Load(); got != 1 {
t.Fatalf("maximum concurrent executions = %d, want 1", got)
}
for _, runID := range []string{first.ID, second.ID} {
events, err := store.ListRunEvents(context.Background(), runID, 0, 20)
if err != nil {
t.Fatalf("list events for %s: %v", runID, err)
}
if len(events) != 3 {
t.Fatalf("event count for %s = %d, want 3: %#v", runID, len(events), events)
}
for index, event := range events {
if event.Sequence != int64(index+1) {
t.Fatalf("event sequence at %d = %d, want %d", index, event.Sequence, index+1)
}
}
}
}
func TestManagerCancelsQueuedAndRunningRuns(t *testing.T) {
store := openTestStore(t)
definition := putTestJob(t, store, "queue")
started := make(chan string, 2)
exited := make(chan struct{}, 1)
executor := ExecutorFunc(func(ctx context.Context, request ExecutionRequest, _ RunReporter) (ExecutionOutcome, error) {
started <- request.Run.ID
<-ctx.Done()
exited <- struct{}{}
return ExecutionOutcome{}, context.Cause(ctx)
})
manager := newTestManager(t, store, executor)
first, err := manager.StartRun(context.Background(), definition.ID)
if err != nil {
t.Fatalf("start first run: %v", err)
}
if got := receiveString(t, started); got != first.ID {
t.Fatalf("executed run = %q, want %q", got, first.ID)
}
second, err := manager.StartRun(context.Background(), definition.ID)
if err != nil {
t.Fatalf("queue second run: %v", err)
}
if err := manager.CancelRun(context.Background(), second.ID); err != nil {
t.Fatalf("cancel queued run: %v", err)
}
waitRunStatus(t, store, second.ID, RunStatusCanceled)
if err := manager.CancelRun(context.Background(), first.ID); err != nil {
t.Fatalf("cancel running run: %v", err)
}
waitRunStatus(t, store, first.ID, RunStatusCanceled)
select {
case <-exited:
case <-time.After(time.Second):
t.Fatal("executor did not observe cancellation")
}
select {
case got := <-started:
t.Fatalf("canceled queued run unexpectedly executed: %s", got)
case <-time.After(75 * time.Millisecond):
}
events, err := store.ListRunEvents(context.Background(), first.ID, 0, 20)
if err != nil {
t.Fatalf("list canceled run events: %v", err)
}
if len(events) < 4 || events[len(events)-2].Type != RunEventCancelling || events[len(events)-1].Type != RunEventCanceled {
t.Fatalf("cancellation event order = %#v", events)
}
}
func TestManagerForbidPolicyRejectsAnUnfinishedRun(t *testing.T) {
store := openTestStore(t)
definition := putTestJob(t, store, "forbid")
started := make(chan string, 1)
executor := ExecutorFunc(func(ctx context.Context, request ExecutionRequest, _ RunReporter) (ExecutionOutcome, error) {
started <- request.Run.ID
<-ctx.Done()
return ExecutionOutcome{}, context.Cause(ctx)
})
manager := newTestManager(t, store, executor)
first, err := manager.StartRun(context.Background(), definition.ID)
if err != nil {
t.Fatalf("start first run: %v", err)
}
if got := receiveString(t, started); got != first.ID {
t.Fatalf("executed run = %q, want %q", got, first.ID)
}
if _, err := manager.StartRun(context.Background(), definition.ID); !errors.Is(err, ErrRunAlreadyActive) {
t.Fatalf("start overlapping run error = %v, want ErrRunAlreadyActive", err)
}
}
func TestManagerResumesAFailedRunFromItsCheckpoint(t *testing.T) {
store := openTestStore(t)
definition := putTestJob(t, store, "forbid")
requests := make(chan ExecutionRequest, 2)
var calls atomic.Int32
executor := ExecutorFunc(func(_ context.Context, request ExecutionRequest, reporter RunReporter) (ExecutionOutcome, error) {
requests <- request
if calls.Add(1) == 1 {
if err := reporter.SaveCheckpoint(Checkpoint{
Kind: "watermark",
Table: "orders",
Phase: "copy",
CursorType: "primary_key",
Cursor: []byte(`{"id":42}`),
}); err != nil {
return ExecutionOutcome{}, err
}
return ExecutionOutcome{RowsInserted: 42, Resumable: true}, errors.New("target unavailable")
}
return ExecutionOutcome{RowsInserted: 1}, nil
})
manager := newTestManager(t, store, executor)
failed, err := manager.StartRun(context.Background(), definition.ID)
if err != nil {
t.Fatalf("start run: %v", err)
}
waitRunStatus(t, store, failed.ID, RunStatusFailed)
resumed, err := manager.ResumeRun(context.Background(), failed.ID)
if err != nil {
t.Fatalf("resume run: %v", err)
}
if resumed.ParentRunID != failed.ID || resumed.Attempt != 2 || resumed.Trigger != RunTriggerResume {
t.Fatalf("resumed run lineage = %#v", resumed)
}
waitRunStatus(t, store, resumed.ID, RunStatusSucceeded)
firstRequest := receiveRequest(t, requests)
secondRequest := receiveRequest(t, requests)
if firstRequest.Checkpoint != nil {
t.Fatalf("initial request unexpectedly received checkpoint: %#v", firstRequest.Checkpoint)
}
if secondRequest.Checkpoint == nil || secondRequest.Checkpoint.RunID != failed.ID || string(secondRequest.Checkpoint.Cursor) != `{"id":42}` {
t.Fatalf("resume checkpoint = %#v", secondRequest.Checkpoint)
}
}
func TestManagerClearsCheckpointAfterSuccessfulSnapshot(t *testing.T) {
store := openTestStore(t)
definition := putTestJob(t, store, "forbid")
executor := ExecutorFunc(func(_ context.Context, _ ExecutionRequest, reporter RunReporter) (ExecutionOutcome, error) {
if err := reporter.SaveCheckpoint(Checkpoint{
Version: 1,
Kind: "resume",
Table: "orders",
Phase: "mapping_completed",
CursorType: "mapping_index",
Cursor: json.RawMessage(`{"nextMapping":1}`),
}); err != nil {
return ExecutionOutcome{}, err
}
return ExecutionOutcome{RowsInserted: 1}, nil
})
manager, err := NewManager(context.Background(), store, executor, ManagerOptions{
SchedulerInterval: time.Hour,
HeartbeatInterval: time.Hour,
})
if err != nil {
t.Fatalf("new manager: %v", err)
}
t.Cleanup(func() {
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
_ = manager.Shutdown(ctx)
})
run, err := manager.StartRun(context.Background(), definition.ID)
if err != nil {
t.Fatalf("start run: %v", err)
}
waitRunStatus(t, store, run.ID, RunStatusSucceeded)
if _, err := store.GetCheckpoint(context.Background(), definition.ID); !errors.Is(err, ErrNotFound) {
t.Fatalf("completed snapshot checkpoint error = %v, want ErrNotFound", err)
}
}
func TestManagerKeepsCheckpointAfterSuccessfulWatermarkRun(t *testing.T) {
store := openTestStore(t)
definition := putTestJob(t, store, "forbid")
definition.IncrementalMode = IncrementalWatermark
definition.Mappings[0].Watermark = &WatermarkSpec{Column: "updated_at", TieBreakerColumns: []string{"id"}}
definition, err := store.PutJob(context.Background(), definition)
if err != nil {
t.Fatalf("update watermark definition: %v", err)
}
executor := ExecutorFunc(func(_ context.Context, request ExecutionRequest, reporter RunReporter) (ExecutionOutcome, error) {
if request.Definition.IncrementalMode != IncrementalWatermark {
return ExecutionOutcome{}, errors.New("executor received non-watermark definition")
}
if err := reporter.SaveCheckpoint(Checkpoint{
Version: 1,
Kind: "watermark",
Table: "orders",
Phase: "batch_committed",
CursorType: "watermark_map",
Cursor: json.RawMessage(`{"orders":{"updatedAt":"2026-08-08T00:00:00Z","id":42}}`),
}); err != nil {
return ExecutionOutcome{}, err
}
return ExecutionOutcome{RowsUpdated: 1}, nil
})
manager := newTestManager(t, store, executor)
run, err := manager.StartRun(context.Background(), definition.ID)
if err != nil {
t.Fatalf("start run: %v", err)
}
waitRunStatus(t, store, run.ID, RunStatusSucceeded)
checkpoint, err := store.GetCheckpoint(context.Background(), definition.ID)
if err != nil {
t.Fatalf("get watermark checkpoint: %v", err)
}
if checkpoint.RunID != run.ID || checkpoint.Kind != "watermark" {
t.Fatalf("unexpected watermark checkpoint: %#v", checkpoint)
}
}
func TestManagerRecoversStaleRunningAndQueuedRunsOnStartup(t *testing.T) {
store := openTestStore(t)
definition := putTestJob(t, store, "queue")
snapshot, err := json.Marshal(definition)
if err != nil {
t.Fatalf("marshal definition: %v", err)
}
stale, err := store.CreateRun(context.Background(), RunRecord{
JobID: definition.ID,
JobRevision: definition.Revision,
Status: RunStatusRunning,
StartedAt: time.Now().Add(-time.Hour).UnixMilli(),
HeartbeatAt: time.Now().Add(-time.Hour).UnixMilli(),
DefinitionSnapshot: snapshot,
})
if err != nil {
t.Fatalf("create stale run: %v", err)
}
queued, err := store.CreateRun(context.Background(), RunRecord{
JobID: definition.ID,
JobRevision: definition.Revision,
Status: RunStatusQueued,
DefinitionSnapshot: snapshot,
})
if err != nil {
t.Fatalf("create queued run: %v", err)
}
executed := make(chan string, 1)
manager, err := NewManager(context.Background(), store, ExecutorFunc(func(_ context.Context, request ExecutionRequest, _ RunReporter) (ExecutionOutcome, error) {
executed <- request.Run.ID
return ExecutionOutcome{}, nil
}), ManagerOptions{
SchedulerInterval: time.Hour,
HeartbeatInterval: time.Hour,
RecoveryStaleAfter: time.Second,
})
if err != nil {
t.Fatalf("new manager: %v", err)
}
t.Cleanup(func() {
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
if err := manager.Shutdown(ctx); err != nil {
t.Errorf("shutdown manager: %v", err)
}
})
recovered := waitRunStatus(t, store, stale.ID, RunStatusInterrupted)
if !recovered.Resumable || recovered.FinishedAt == 0 {
t.Fatalf("recovered stale run = %#v", recovered)
}
if got := receiveString(t, executed); got != queued.ID {
t.Fatalf("restored queued run = %q, want %q", got, queued.ID)
}
waitRunStatus(t, store, queued.ID, RunStatusSucceeded)
events, err := store.ListRunEvents(context.Background(), stale.ID, 0, 10)
if err != nil {
t.Fatalf("list stale run events: %v", err)
}
if len(events) != 1 || events[0].Type != RunEventInterrupted {
t.Fatalf("stale run events = %#v", events)
}
}
func TestManagersUseSQLiteLeaseToScheduleOneRun(t *testing.T) {
databasePath := t.TempDir() + "/shared-sync-jobs.db"
firstStore, err := Open(databasePath)
if err != nil {
t.Fatalf("open first store: %v", err)
}
t.Cleanup(func() { _ = firstStore.Close() })
secondStore, err := Open(databasePath)
if err != nil {
t.Fatalf("open second store: %v", err)
}
t.Cleanup(func() { _ = secondStore.Close() })
definition, err := firstStore.PutJob(context.Background(), JobDefinition{
Name: "scheduled orders sync",
Enabled: true,
Kind: JobKindReconcile,
IncrementalMode: IncrementalSnapshot,
Source: EndpointRef{ConnectionID: "source"},
Target: EndpointRef{ConnectionID: "target"},
Mappings: []TableMapping{{SourceTable: "orders", TargetTable: "orders", Enabled: true}},
ConcurrencyPolicy: "queue",
Schedule: ScheduleSpec{
Kind: ScheduleInterval,
IntervalSeconds: 10,
MisfirePolicy: "run_once",
},
})
if err != nil {
t.Fatalf("put scheduled job: %v", err)
}
dueAt := time.Now().Add(-time.Second).UnixMilli()
if _, err := firstStore.db.ExecContext(context.Background(), `UPDATE data_sync_jobs SET next_run_at = ? WHERE id = ?`, dueAt, definition.ID); err != nil {
t.Fatalf("make job due: %v", err)
}
var executions atomic.Int32
executor := ExecutorFunc(func(context.Context, ExecutionRequest, RunReporter) (ExecutionOutcome, error) {
executions.Add(1)
return ExecutionOutcome{}, nil
})
firstManager := newScheduledTestManager(t, firstStore, executor, "owner-a")
_ = firstManager
secondManager := newScheduledTestManager(t, secondStore, executor, "owner-b")
_ = secondManager
deadline := time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) {
runs, listErr := firstStore.ListRuns(context.Background(), definition.ID, 10)
if listErr == nil && len(runs) == 1 && runs[0].Status == RunStatusSucceeded {
if runs[0].Trigger != RunTriggerSchedule {
t.Fatalf("scheduled run trigger = %q", runs[0].Trigger)
}
break
}
time.Sleep(10 * time.Millisecond)
}
time.Sleep(100 * time.Millisecond)
runs, err := firstStore.ListRuns(context.Background(), definition.ID, 10)
if err != nil {
t.Fatalf("list scheduled runs: %v", err)
}
if len(runs) != 1 || executions.Load() != 1 {
t.Fatalf("scheduled runs = %d, executions = %d; want one", len(runs), executions.Load())
}
}
func TestManagerShutdownCancelsExecutorsAndWaitsForGoroutines(t *testing.T) {
store := openTestStore(t)
definition := putTestJob(t, store, "queue")
started := make(chan struct{})
exited := make(chan struct{})
executor := ExecutorFunc(func(ctx context.Context, _ ExecutionRequest, _ RunReporter) (ExecutionOutcome, error) {
close(started)
<-ctx.Done()
time.Sleep(25 * time.Millisecond)
close(exited)
return ExecutionOutcome{Resumable: true}, context.Cause(ctx)
})
manager, err := NewManager(context.Background(), store, executor, ManagerOptions{
SchedulerInterval: time.Hour,
HeartbeatInterval: time.Hour,
})
if err != nil {
t.Fatalf("new manager: %v", err)
}
run, err := manager.StartRun(context.Background(), definition.ID)
if err != nil {
t.Fatalf("start run: %v", err)
}
select {
case <-started:
case <-time.After(time.Second):
t.Fatal("executor did not start")
}
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
if err := manager.Shutdown(ctx); err != nil {
t.Fatalf("shutdown manager: %v", err)
}
select {
case <-exited:
default:
t.Fatal("shutdown returned before executor goroutine exited")
}
recovered := waitRunStatus(t, store, run.ID, RunStatusInterrupted)
if !recovered.Resumable {
t.Fatalf("shutdown-interrupted run is not resumable: %#v", recovered)
}
}
func TestManagerAutomaticallyResumesRecoveredRunWhenConfigured(t *testing.T) {
store := openTestStore(t)
definition := putTestJob(t, store, "forbid")
definition.ResumePolicy = "auto"
definition, err := store.PutJob(context.Background(), definition)
if err != nil {
t.Fatalf("enable auto resume: %v", err)
}
snapshot, err := json.Marshal(definition)
if err != nil {
t.Fatalf("marshal definition: %v", err)
}
stale, err := store.CreateRun(context.Background(), RunRecord{
JobID: definition.ID,
JobRevision: definition.Revision,
Status: RunStatusRunning,
StartedAt: time.Now().Add(-time.Hour).UnixMilli(),
HeartbeatAt: time.Now().Add(-time.Hour).UnixMilli(),
DefinitionSnapshot: snapshot,
})
if err != nil {
t.Fatalf("create stale run: %v", err)
}
if _, err := store.PutCheckpoint(context.Background(), Checkpoint{
Kind: "watermark",
JobID: definition.ID,
RunID: stale.ID,
DefinitionRevision: definition.Revision,
Table: "orders",
Phase: "copy",
CursorType: "primary_key",
Cursor: []byte(`{"id":99}`),
}); err != nil {
t.Fatalf("put checkpoint: %v", err)
}
requests := make(chan ExecutionRequest, 1)
manager, err := NewManager(context.Background(), store, ExecutorFunc(func(_ context.Context, request ExecutionRequest, _ RunReporter) (ExecutionOutcome, error) {
requests <- request
return ExecutionOutcome{}, nil
}), ManagerOptions{
SchedulerInterval: time.Hour,
HeartbeatInterval: time.Hour,
RecoveryStaleAfter: time.Second,
})
if err != nil {
t.Fatalf("new manager: %v", err)
}
t.Cleanup(func() {
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
_ = manager.Shutdown(ctx)
})
request := receiveRequest(t, requests)
if request.Run.Trigger != RunTriggerResume || request.Run.ParentRunID != stale.ID || request.Checkpoint == nil || request.Checkpoint.RunID != stale.ID {
t.Fatalf("automatic resume request = %#v", request)
}
waitRunStatus(t, store, request.Run.ID, RunStatusSucceeded)
}
func TestManagerPersistsReporterOutputBeforePublishingHooks(t *testing.T) {
store := openTestStore(t)
definition := putTestJob(t, store, "queue")
var hookMu sync.Mutex
hooked := make([]RunEvent, 0)
executor := ExecutorFunc(func(_ context.Context, _ ExecutionRequest, reporter RunReporter) (ExecutionOutcome, error) {
if err := reporter.ReportProgress(RunProgress{Current: 3, Total: 5, Table: "orders", Stage: "write", Message: "batch 3"}); err != nil {
return ExecutionOutcome{}, err
}
if err := reporter.AppendErrorRow(ErrorRow{Error: "duplicate key", SourceTable: "orders", TargetTable: "orders", SourceKey: []byte(`{"id":3}`)}); err != nil {
return ExecutionOutcome{}, err
}
if err := reporter.Emit(RunEventLog, "executor log", []byte(`{"level":"info"}`)); err != nil {
return ExecutionOutcome{}, err
}
return ExecutionOutcome{RowsInserted: 2, RowsFailed: 1, Message: "completed with errors"}, nil
})
manager, err := NewManager(context.Background(), store, executor, ManagerOptions{
SchedulerInterval: time.Hour,
HeartbeatInterval: time.Hour,
Hooks: ManagerHooks{OnRunEvent: func(event RunEvent) {
persisted, listErr := store.ListRunEvents(context.Background(), event.RunID, event.Sequence-1, 1)
if listErr != nil || len(persisted) != 1 || persisted[0].Sequence != event.Sequence {
t.Errorf("hook observed event before persistence: event=%#v persisted=%#v err=%v", event, persisted, listErr)
}
hookMu.Lock()
hooked = append(hooked, event)
hookMu.Unlock()
}},
})
if err != nil {
t.Fatalf("new manager: %v", err)
}
t.Cleanup(func() {
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
_ = manager.Shutdown(ctx)
})
run, err := manager.StartRun(context.Background(), definition.ID)
if err != nil {
t.Fatalf("start run: %v", err)
}
completed := waitRunStatus(t, store, run.ID, RunStatusPartial)
if completed.Current != 3 || completed.Total != 5 || completed.RowsInserted != 2 || completed.RowsFailed != 1 {
t.Fatalf("persisted run output = %#v", completed)
}
errorRows, err := store.ListErrorRows(context.Background(), run.ID, ErrorRowPending, 10)
if err != nil || len(errorRows) != 1 || errorRows[0].Error != "duplicate key" {
t.Fatalf("persisted error rows = %#v, err=%v", errorRows, err)
}
wantTypes := []RunEventType{RunEventQueued, RunEventStarted, RunEventProgress, RunEventErrorRow, RunEventLog, RunEventPartial}
events := waitRunEventCount(t, store, run.ID, len(wantTypes))
for index, want := range wantTypes {
if events[index].Type != want {
t.Fatalf("event %d type = %q, want %q", index, events[index].Type, want)
}
}
var hookCount int
deadline := time.Now().Add(time.Second)
for time.Now().Before(deadline) {
hookMu.Lock()
hookCount = len(hooked)
hookMu.Unlock()
if hookCount == len(events) {
break
}
time.Sleep(time.Millisecond)
}
if hookCount != len(events) {
t.Fatalf("hook event count = %d, persisted = %d", hookCount, len(events))
}
}
func TestManagerRetriesTerminalRunWithOriginalSnapshotAndLineage(t *testing.T) {
store := openTestStore(t)
definition := putTestJob(t, store, "forbid")
requests := make(chan ExecutionRequest, 2)
var calls atomic.Int32
executor := ExecutorFunc(func(_ context.Context, request ExecutionRequest, _ RunReporter) (ExecutionOutcome, error) {
requests <- request
if calls.Add(1) == 1 {
return ExecutionOutcome{}, errors.New("temporary target failure")
}
return ExecutionOutcome{}, nil
})
manager := newTestManager(t, store, executor)
failed, err := manager.StartRun(context.Background(), definition.ID)
if err != nil {
t.Fatalf("start run: %v", err)
}
waitRunStatus(t, store, failed.ID, RunStatusFailed)
retried, err := manager.RetryRun(context.Background(), failed.ID)
if err != nil {
t.Fatalf("retry failed run: %v", err)
}
if retried.Trigger != RunTriggerRetry || retried.ParentRunID != failed.ID || retried.Attempt != failed.Attempt+1 {
t.Fatalf("retry lineage = %#v", retried)
}
waitRunStatus(t, store, retried.ID, RunStatusSucceeded)
_ = receiveRequest(t, requests)
retryRequest := receiveRequest(t, requests)
if retryRequest.Definition.ID != definition.ID || retryRequest.Definition.Revision != definition.Revision {
t.Fatalf("retry definition snapshot = %#v", retryRequest.Definition)
}
if _, err := manager.RetryRun(context.Background(), retried.ID); !errors.Is(err, ErrRunNotRetryable) {
t.Fatalf("retry succeeded run error = %v, want ErrRunNotRetryable", err)
}
}
func TestManagerRejectsRetryWhenOriginalSnapshotDoesNotMatchRun(t *testing.T) {
store := openTestStore(t)
definition := putTestJob(t, store, "forbid")
snapshot, err := json.Marshal(definition)
if err != nil {
t.Fatalf("marshal definition: %v", err)
}
run, err := store.CreateRun(context.Background(), RunRecord{
JobID: definition.ID,
JobRevision: definition.Revision + 1,
Status: RunStatusFailed,
DefinitionSnapshot: snapshot,
})
if err != nil {
t.Fatalf("create inconsistent run: %v", err)
}
manager := newTestManager(t, store, ExecutorFunc(func(context.Context, ExecutionRequest, RunReporter) (ExecutionOutcome, error) {
t.Fatal("inconsistent retry must not execute")
return ExecutionOutcome{}, nil
}))
if _, err := manager.RetryRun(context.Background(), run.ID); err == nil || errors.Is(err, ErrRunNotRetryable) {
t.Fatalf("inconsistent retry error = %v, want snapshot consistency error", err)
}
}
func TestManagerRetriesAfterMetadataOnlyTaskRevisionChanges(t *testing.T) {
store := openTestStore(t)
definition := putTestJob(t, store, "forbid")
manager := newTestManager(t, store, ExecutorFunc(func(context.Context, ExecutionRequest, RunReporter) (ExecutionOutcome, error) {
return ExecutionOutcome{}, errors.New("temporary target failure")
}))
failed, err := manager.StartRun(context.Background(), definition.ID)
if err != nil {
t.Fatalf("start failing run: %v", err)
}
waitRunStatus(t, store, failed.ID, RunStatusFailed)
definition.Name = "renamed task"
if _, err := store.PutJob(context.Background(), definition); err != nil {
t.Fatalf("update task: %v", err)
}
updated, err := store.GetJob(context.Background(), definition.ID)
if err != nil {
t.Fatalf("get renamed task: %v", err)
}
retried, err := manager.RetryRun(context.Background(), failed.ID)
if err != nil {
t.Fatalf("retry after metadata change: %v", err)
}
if retried.JobRevision != updated.Revision {
t.Fatalf("retry revision = %d, want current %d", retried.JobRevision, updated.Revision)
}
}
func TestManagerRejectsRetryAfterExecutionPlanChanges(t *testing.T) {
store := openTestStore(t)
definition := putTestJob(t, store, "forbid")
manager := newTestManager(t, store, ExecutorFunc(func(context.Context, ExecutionRequest, RunReporter) (ExecutionOutcome, error) {
return ExecutionOutcome{}, errors.New("temporary target failure")
}))
failed, err := manager.StartRun(context.Background(), definition.ID)
if err != nil {
t.Fatalf("start failing run: %v", err)
}
waitRunStatus(t, store, failed.ID, RunStatusFailed)
definition.Mappings[0].TargetTable = "orders_v2"
if _, err := store.PutJob(context.Background(), definition); err != nil {
t.Fatalf("update task plan: %v", err)
}
if _, err := manager.RetryRun(context.Background(), failed.ID); !errors.Is(err, ErrRevisionConflict) {
t.Fatalf("retry changed plan error = %v, want ErrRevisionConflict", err)
}
}
func TestManagerRetryObeysForbidConcurrencyPolicy(t *testing.T) {
store := openTestStore(t)
definition := putTestJob(t, store, "forbid")
started := make(chan string, 1)
var calls atomic.Int32
executor := ExecutorFunc(func(ctx context.Context, request ExecutionRequest, _ RunReporter) (ExecutionOutcome, error) {
if calls.Add(1) == 1 {
return ExecutionOutcome{}, errors.New("first run failed")
}
started <- request.Run.ID
<-ctx.Done()
return ExecutionOutcome{}, context.Cause(ctx)
})
manager := newTestManager(t, store, executor)
failed, err := manager.StartRun(context.Background(), definition.ID)
if err != nil {
t.Fatalf("start failing run: %v", err)
}
waitRunStatus(t, store, failed.ID, RunStatusFailed)
blocker, err := manager.StartRun(context.Background(), definition.ID)
if err != nil {
t.Fatalf("start blocking run: %v", err)
}
if got := receiveString(t, started); got != blocker.ID {
t.Fatalf("blocking run = %q, want %q", got, blocker.ID)
}
if _, err := manager.RetryRun(context.Background(), failed.ID); !errors.Is(err, ErrRunAlreadyActive) {
t.Fatalf("retry during active run error = %v, want ErrRunAlreadyActive", err)
}
if err := manager.CancelRun(context.Background(), blocker.ID); err != nil {
t.Fatalf("cancel blocking run: %v", err)
}
waitRunStatus(t, store, blocker.ID, RunStatusCanceled)
}
func openTestStore(t *testing.T) *Store {
t.Helper()
store, err := Open(t.TempDir() + "/sync-jobs.db")
if err != nil {
t.Fatalf("open store: %v", err)
}
t.Cleanup(func() {
if err := store.Close(); err != nil {
t.Errorf("close store: %v", err)
}
})
return store
}
func newTestManager(t *testing.T, store *Store, executor Executor) *Manager {
t.Helper()
manager, err := NewManager(context.Background(), store, executor, ManagerOptions{
SchedulerInterval: time.Hour,
HeartbeatInterval: time.Hour,
})
if err != nil {
t.Fatalf("new manager: %v", err)
}
t.Cleanup(func() {
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
if err := manager.Shutdown(ctx); err != nil {
t.Errorf("shutdown manager: %v", err)
}
})
return manager
}
func newScheduledTestManager(t *testing.T, store *Store, executor Executor, owner string) *Manager {
t.Helper()
manager, err := NewManager(context.Background(), store, executor, ManagerOptions{
SchedulerInterval: 10 * time.Millisecond,
LeaseTTL: 100 * time.Millisecond,
HeartbeatInterval: time.Hour,
RecoveryStaleAfter: time.Hour,
LeaseOwner: owner,
})
if err != nil {
t.Fatalf("new scheduled manager: %v", err)
}
t.Cleanup(func() {
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
if err := manager.Shutdown(ctx); err != nil {
t.Errorf("shutdown scheduled manager: %v", err)
}
})
return manager
}
func putTestJob(t *testing.T, store *Store, concurrencyPolicy string) JobDefinition {
t.Helper()
definition, err := store.PutJob(context.Background(), JobDefinition{
Name: "orders sync",
Enabled: true,
Kind: JobKindReconcile,
IncrementalMode: IncrementalSnapshot,
Source: EndpointRef{ConnectionID: "source"},
Target: EndpointRef{ConnectionID: "target"},
Mappings: []TableMapping{{SourceTable: "orders", TargetTable: "orders", Enabled: true}},
ConcurrencyPolicy: concurrencyPolicy,
})
if err != nil {
t.Fatalf("put test job: %v", err)
}
return definition
}
func receiveString(t *testing.T, values <-chan string) string {
t.Helper()
select {
case value := <-values:
return value
case <-time.After(2 * time.Second):
t.Fatal("timed out waiting for value")
return ""
}
}
func receiveRequest(t *testing.T, values <-chan ExecutionRequest) ExecutionRequest {
t.Helper()
select {
case value := <-values:
return value
case <-time.After(2 * time.Second):
t.Fatal("timed out waiting for execution request")
return ExecutionRequest{}
}
}
func assertRunStatus(t *testing.T, store *Store, runID string, want RunStatus) {
t.Helper()
run, err := store.GetRun(context.Background(), runID)
if err != nil {
t.Fatalf("get run %s: %v", runID, err)
}
if run.Status != want {
t.Fatalf("run %s status = %q, want %q", runID, run.Status, want)
}
}
func waitRunStatus(t *testing.T, store *Store, runID string, want RunStatus) RunRecord {
t.Helper()
deadline := time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) {
run, err := store.GetRun(context.Background(), runID)
if err == nil && run.Status == want {
return run
}
time.Sleep(5 * time.Millisecond)
}
assertRunStatus(t, store, runID, want)
return RunRecord{}
}
func waitRunEventCount(t *testing.T, store *Store, runID string, want int) []RunEvent {
t.Helper()
deadline := time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) {
events, err := store.ListRunEvents(context.Background(), runID, 0, want+10)
if err == nil && len(events) >= want {
return events
}
time.Sleep(5 * time.Millisecond)
}
events, err := store.ListRunEvents(context.Background(), runID, 0, want+10)
if err != nil {
t.Fatalf("list run events: %v", err)
}
t.Fatalf("event count = %d, want at least %d: %#v", len(events), want, events)
return nil
}

306
internal/syncjob/model.go Normal file
View File

@@ -0,0 +1,306 @@
package syncjob
import "encoding/json"
const CurrentDefinitionVersion = 1
type JobLifecycle string
const (
JobLifecycleDraft JobLifecycle = "draft"
JobLifecycleReady JobLifecycle = "ready"
JobLifecycleEnabled JobLifecycle = "enabled"
JobLifecyclePaused JobLifecycle = "paused"
JobLifecycleArchived JobLifecycle = "archived"
)
type JobKind string
const (
JobKindMigration JobKind = "migration"
JobKindReconcile JobKind = "reconcile"
JobKindQuerySink JobKind = "query_sink"
JobKindCompare JobKind = "compare"
)
type IncrementalMode string
const (
IncrementalSnapshot IncrementalMode = "snapshot"
IncrementalWatermark IncrementalMode = "watermark"
IncrementalCDC IncrementalMode = "cdc"
)
type ScheduleKind string
const (
ScheduleManual ScheduleKind = "manual"
ScheduleOnce ScheduleKind = "once"
ScheduleInterval ScheduleKind = "interval"
ScheduleCron ScheduleKind = "cron"
ScheduleContinuous ScheduleKind = "continuous"
)
type ErrorPolicy string
const (
ErrorPolicyStop ErrorPolicy = "stop"
ErrorPolicySkipRow ErrorPolicy = "skip_row"
)
type EndpointRef struct {
ConnectionID string `json:"connectionId"`
ConnectionType string `json:"connectionType,omitempty"`
ConnectionName string `json:"connectionName,omitempty"`
Database string `json:"database,omitempty"`
Schema string `json:"schema,omitempty"`
Fingerprint string `json:"fingerprint,omitempty"`
}
type ExecutionApproval struct {
DefinitionHash string `json:"definitionHash"`
TargetFingerprint string `json:"targetFingerprint"`
ApprovedAt int64 `json:"approvedAt"`
ApprovedByRuntime string `json:"approvedByRuntime"`
}
type TransformSpec struct {
Kind string `json:"kind,omitempty"`
Argument json.RawMessage `json:"argument,omitempty"`
}
type ColumnMapping struct {
Source string `json:"source,omitempty"`
Target string `json:"target"`
Transform TransformSpec `json:"transform,omitempty"`
DefaultValue json.RawMessage `json:"defaultValue,omitempty"`
Required bool `json:"required,omitempty"`
}
type WatermarkSpec struct {
Column string `json:"column"`
InitialValue json.RawMessage `json:"initialValue,omitempty"`
TieBreakerColumns []string `json:"tieBreakerColumns,omitempty"`
}
type TableMapping struct {
SourceSchema string `json:"sourceSchema,omitempty"`
SourceTable string `json:"sourceTable"`
TargetSchema string `json:"targetSchema,omitempty"`
TargetTable string `json:"targetTable"`
TargetTableStrategy string `json:"targetTableStrategy,omitempty"`
Filter string `json:"filter,omitempty"`
KeyColumns []string `json:"keyColumns,omitempty"`
Columns []ColumnMapping `json:"columns,omitempty"`
Watermark *WatermarkSpec `json:"watermark,omitempty"`
Enabled bool `json:"enabled"`
}
type ExecutionOptions struct {
Content string `json:"content,omitempty"`
SyncMode string `json:"syncMode,omitempty"`
TargetTableStrategy string `json:"targetTableStrategy,omitempty"`
AutoAddColumns bool `json:"autoAddColumns,omitempty"`
CreateIndexes bool `json:"createIndexes,omitempty"`
PropagateDeletes bool `json:"propagateDeletes,omitempty"`
BatchSize int `json:"batchSize,omitempty"`
ErrorPolicy ErrorPolicy `json:"errorPolicy,omitempty"`
MaxRetries int `json:"maxRetries,omitempty"`
RetryBackoffMillis int `json:"retryBackoffMillis,omitempty"`
CaptureErrorPayload bool `json:"captureErrorPayload,omitempty"`
}
type ScheduleSpec struct {
Kind ScheduleKind `json:"kind"`
RunAt int64 `json:"runAt,omitempty"`
IntervalSeconds int64 `json:"intervalSeconds,omitempty"`
CronExpression string `json:"cronExpression,omitempty"`
Timezone string `json:"timezone,omitempty"`
AnchorAt int64 `json:"anchorAt,omitempty"`
MisfirePolicy string `json:"misfirePolicy,omitempty"`
}
type CDCSpec struct {
Adapter string `json:"adapter,omitempty"`
StartPosition string `json:"startPosition,omitempty"`
InitialSnapshot bool `json:"initialSnapshot,omitempty"`
SlotName string `json:"slotName,omitempty"`
PublicationName string `json:"publicationName,omitempty"`
}
type JobDefinition struct {
Version int `json:"version"`
ID string `json:"id"`
Name string `json:"name"`
Description string `json:"description,omitempty"`
Lifecycle JobLifecycle `json:"lifecycle"`
Enabled bool `json:"enabled"`
Kind JobKind `json:"kind"`
IncrementalMode IncrementalMode `json:"incrementalMode"`
Source EndpointRef `json:"source"`
Target EndpointRef `json:"target"`
SourceQuery string `json:"sourceQuery,omitempty"`
Mappings []TableMapping `json:"mappings"`
Options ExecutionOptions `json:"options"`
Schedule ScheduleSpec `json:"schedule"`
CDC *CDCSpec `json:"cdc,omitempty"`
Approval *ExecutionApproval `json:"approval,omitempty"`
ConcurrencyPolicy string `json:"concurrencyPolicy,omitempty"`
ResumePolicy string `json:"resumePolicy,omitempty"`
Revision int64 `json:"revision"`
CreatedAt int64 `json:"createdAt"`
UpdatedAt int64 `json:"updatedAt"`
NextRunAt int64 `json:"nextRunAt,omitempty"`
LastScheduledAt int64 `json:"lastScheduledAt,omitempty"`
ArchivedAt int64 `json:"archivedAt,omitempty"`
}
type RunStatus string
const (
RunStatusQueued RunStatus = "queued"
RunStatusRunning RunStatus = "running"
RunStatusCancelling RunStatus = "cancelling"
RunStatusPaused RunStatus = "paused"
RunStatusSucceeded RunStatus = "succeeded"
RunStatusPartial RunStatus = "partial"
RunStatusFailed RunStatus = "failed"
RunStatusCanceled RunStatus = "canceled"
RunStatusInterrupted RunStatus = "interrupted"
)
type RunTrigger string
const (
RunTriggerManual RunTrigger = "manual"
RunTriggerSchedule RunTrigger = "schedule"
RunTriggerResume RunTrigger = "resume"
RunTriggerRetry RunTrigger = "retry"
)
type RunRecord struct {
ID string `json:"id"`
JobID string `json:"jobId"`
OwnerToken string `json:"-"`
JobRevision int64 `json:"jobRevision"`
Trigger RunTrigger `json:"trigger"`
Status RunStatus `json:"status"`
ParentRunID string `json:"parentRunId,omitempty"`
Attempt int `json:"attempt"`
QueuedAt int64 `json:"queuedAt"`
StartedAt int64 `json:"startedAt,omitempty"`
FinishedAt int64 `json:"finishedAt,omitempty"`
HeartbeatAt int64 `json:"heartbeatAt,omitempty"`
Current int `json:"current"`
Total int `json:"total"`
Table string `json:"table,omitempty"`
Stage string `json:"stage,omitempty"`
RowsInserted int64 `json:"rowsInserted"`
RowsUpdated int64 `json:"rowsUpdated"`
RowsDeleted int64 `json:"rowsDeleted"`
RowsFailed int64 `json:"rowsFailed"`
Message string `json:"message,omitempty"`
Resumable bool `json:"resumable"`
DefinitionSnapshot json.RawMessage `json:"definitionSnapshot,omitempty"`
SourceFingerprint string `json:"sourceFingerprint,omitempty"`
TargetFingerprint string `json:"targetFingerprint,omitempty"`
CreatedAt int64 `json:"createdAt"`
UpdatedAt int64 `json:"updatedAt"`
}
type Checkpoint struct {
Version int `json:"version"`
Kind string `json:"kind"`
JobID string `json:"jobId"`
RunID string `json:"runId"`
DefinitionRevision int64 `json:"definitionRevision"`
Table string `json:"table"`
Phase string `json:"phase"`
CursorType string `json:"cursorType"`
Cursor json.RawMessage `json:"cursor,omitempty"`
Watermark json.RawMessage `json:"watermark,omitempty"`
BatchSequence int64 `json:"batchSequence"`
SchemaHash string `json:"schemaHash,omitempty"`
UpdatedAt int64 `json:"updatedAt"`
}
type ErrorRowStatus string
const (
ErrorRowPending ErrorRowStatus = "pending"
ErrorRowRetrying ErrorRowStatus = "retrying"
ErrorRowResolved ErrorRowStatus = "resolved"
ErrorRowDiscarded ErrorRowStatus = "discarded"
)
type ErrorRow struct {
ID string `json:"id"`
RunID string `json:"runId"`
JobID string `json:"jobId"`
SourceTable string `json:"sourceTable,omitempty"`
TargetTable string `json:"targetTable,omitempty"`
Operation string `json:"operation,omitempty"`
SourceKey json.RawMessage `json:"sourceKey,omitempty"`
Payload json.RawMessage `json:"payload,omitempty"`
PayloadPolicy string `json:"payloadPolicy,omitempty"`
PayloadHash string `json:"payloadHash,omitempty"`
PayloadSize int64 `json:"payloadSize,omitempty"`
Error string `json:"error"`
ErrorCode string `json:"errorCode,omitempty"`
ErrorClass string `json:"errorClass,omitempty"`
Attempts int `json:"attempts"`
Status ErrorRowStatus `json:"status"`
RetryOwner string `json:"-"`
RetryLeaseExpiresAt int64 `json:"-"`
CreatedAt int64 `json:"createdAt"`
UpdatedAt int64 `json:"updatedAt"`
}
type ExecutionOutcome struct {
RowsInserted int64 `json:"rowsInserted"`
RowsUpdated int64 `json:"rowsUpdated"`
RowsDeleted int64 `json:"rowsDeleted"`
RowsFailed int64 `json:"rowsFailed"`
Message string `json:"message,omitempty"`
Resumable bool `json:"resumable"`
}
type RunProgress struct {
Current int `json:"current"`
Total int `json:"total"`
Table string `json:"table,omitempty"`
Stage string `json:"stage,omitempty"`
Message string `json:"message,omitempty"`
}
type RunEventType string
const (
RunEventQueued RunEventType = "queued"
RunEventStarted RunEventType = "started"
RunEventProgress RunEventType = "progress"
RunEventCheckpoint RunEventType = "checkpoint"
RunEventErrorRow RunEventType = "error_row"
RunEventLog RunEventType = "log"
RunEventCancelling RunEventType = "cancelling"
RunEventCanceled RunEventType = "canceled"
RunEventSucceeded RunEventType = "succeeded"
RunEventPartial RunEventType = "partial"
RunEventFailed RunEventType = "failed"
RunEventInterrupted RunEventType = "interrupted"
)
type RunEvent struct {
RunID string `json:"runId"`
JobID string `json:"jobId"`
Sequence int64 `json:"sequence"`
Type RunEventType `json:"type"`
Status RunStatus `json:"status,omitempty"`
Current int `json:"current,omitempty"`
Total int `json:"total,omitempty"`
Table string `json:"table,omitempty"`
Stage string `json:"stage,omitempty"`
Message string `json:"message,omitempty"`
Payload json.RawMessage `json:"payload,omitempty"`
CreatedAt int64 `json:"createdAt"`
}

View File

@@ -0,0 +1,29 @@
package syncjob
// PermanentExecutionError marks an execution failure that cannot be repaired by
// retrying the same persisted task definition. The manager pauses the task only
// when the owning executor successfully commits this run's failed terminal state.
type PermanentExecutionError struct {
Err error
}
func (e *PermanentExecutionError) Error() string {
if e == nil || e.Err == nil {
return "permanent data sync execution failure"
}
return e.Err.Error()
}
func (e *PermanentExecutionError) Unwrap() error {
if e == nil {
return nil
}
return e.Err
}
func MarkPermanentExecutionError(err error) error {
if err == nil {
return nil
}
return &PermanentExecutionError{Err: err}
}

View File

@@ -0,0 +1,63 @@
package syncjob
import (
"crypto/sha256"
"encoding/hex"
"encoding/json"
)
// ExecutionPlanHash identifies the data-affecting portion of a definition.
// Task labels, lifecycle, scheduling and revisions deliberately do not move a
// durable cursor; endpoints, mappings, write policy and incremental scope do.
func ExecutionPlanHash(input JobDefinition) (string, error) {
definition := NormalizeDefinition(input)
definition.ID = ""
definition.Name = ""
definition.Description = ""
definition.Lifecycle = ""
definition.Enabled = false
definition.Source.ConnectionName = ""
definition.Target.ConnectionName = ""
definition.Schedule = ScheduleSpec{}
definition.Approval = nil
definition.ConcurrencyPolicy = ""
definition.ResumePolicy = ""
definition.Revision = 0
definition.CreatedAt = 0
definition.UpdatedAt = 0
definition.NextRunAt = 0
definition.LastScheduledAt = 0
definition.ArchivedAt = 0
return hashJobDefinition(definition)
}
// ApprovalScopeHash identifies the exact operation a user approved. Unlike
// ExecutionPlanHash it deliberately retains task identity, lifecycle,
// scheduling, concurrency and resume policy: approving a manual run must not
// authorize turning the same data plan into an unattended continuous job.
// Only presentation fields, persistence metadata and the approval itself are
// excluded.
func ApprovalScopeHash(input JobDefinition) (string, error) {
definition := NormalizeDefinition(input)
definition.Name = ""
definition.Description = ""
definition.Source.ConnectionName = ""
definition.Target.ConnectionName = ""
definition.Approval = nil
definition.Revision = 0
definition.CreatedAt = 0
definition.UpdatedAt = 0
definition.NextRunAt = 0
definition.LastScheduledAt = 0
definition.ArchivedAt = 0
return hashJobDefinition(definition)
}
func hashJobDefinition(definition JobDefinition) (string, error) {
payload, err := json.Marshal(definition)
if err != nil {
return "", err
}
sum := sha256.Sum256(payload)
return "sha256:" + hex.EncodeToString(sum[:]), nil
}

View File

@@ -0,0 +1,90 @@
package syncjob
import "testing"
func TestExecutionPlanHashIgnoresTaskMetadataButTracksDataSemantics(t *testing.T) {
definition := validValidationTestDefinition()
definition.ID = "job-1"
definition.Revision = 4
definition.Source.Fingerprint = "source-fingerprint"
definition.Target.Fingerprint = "target-fingerprint"
base, err := ExecutionPlanHash(definition)
if err != nil {
t.Fatalf("hash base plan: %v", err)
}
metadata := definition
metadata.Name = "renamed"
metadata.Description = "new description"
metadata.Lifecycle = JobLifecyclePaused
metadata.Enabled = false
metadata.Schedule = ScheduleSpec{Kind: ScheduleCron, CronExpression: "0 1 * * *", Timezone: "UTC"}
metadata.Revision++
got, err := ExecutionPlanHash(metadata)
if err != nil {
t.Fatalf("hash metadata plan: %v", err)
}
if got != base {
t.Fatal("task metadata unexpectedly invalidated the execution plan")
}
changed := definition
changed.Mappings[0].TargetTable = "orders_v2"
got, err = ExecutionPlanHash(changed)
if err != nil {
t.Fatalf("hash changed plan: %v", err)
}
if got == base {
t.Fatal("target mapping change did not invalidate the execution plan")
}
}
func TestApprovalScopeHashTracksTaskAndUnattendedExecutionPolicy(t *testing.T) {
definition := validValidationTestDefinition()
definition.ID = "job-1"
definition.Lifecycle = JobLifecycleReady
definition.Enabled = false
definition.Schedule = ScheduleSpec{Kind: ScheduleManual}
base, err := ApprovalScopeHash(definition)
if err != nil {
t.Fatalf("hash approval scope: %v", err)
}
metadata := definition
metadata.Name = "renamed"
metadata.Description = "new description"
metadata.Revision = 99
metadata.UpdatedAt = 1234
got, err := ApprovalScopeHash(metadata)
if err != nil {
t.Fatalf("hash approval metadata: %v", err)
}
if got != base {
t.Fatal("presentation or persistence metadata unexpectedly invalidated approval")
}
for name, mutate := range map[string]func(*JobDefinition){
"task identity": func(value *JobDefinition) { value.ID = "job-2" },
"lifecycle": func(value *JobDefinition) {
value.Lifecycle = JobLifecycleEnabled
value.Enabled = true
},
"schedule": func(value *JobDefinition) {
value.Schedule = ScheduleSpec{Kind: ScheduleContinuous}
},
"concurrency": func(value *JobDefinition) { value.ConcurrencyPolicy = "queue" },
"resume": func(value *JobDefinition) { value.ResumePolicy = "auto" },
} {
t.Run(name, func(t *testing.T) {
changed := definition
mutate(&changed)
got, hashErr := ApprovalScopeHash(changed)
if hashErr != nil {
t.Fatalf("hash changed approval scope: %v", hashErr)
}
if got == base {
t.Fatalf("%s change did not invalidate approval", name)
}
})
}
}

View File

@@ -0,0 +1,673 @@
package syncjob
import (
"context"
"database/sql"
"encoding/json"
"errors"
"sync/atomic"
"testing"
"time"
)
func TestStoreClaimRunRequiresRunnableJobLifecycle(t *testing.T) {
store := openTestStore(t)
enabled := putTestJob(t, store, "queue")
ready := putLifecycleTestJob(t, store, "ready", JobLifecycleReady)
draft, err := store.PutJob(context.Background(), JobDefinition{Name: "draft", Lifecycle: JobLifecycleDraft})
if err != nil {
t.Fatalf("put draft job: %v", err)
}
paused := putLifecycleTestJob(t, store, "paused", JobLifecycleReady)
paused, err = store.PauseJob(context.Background(), paused.ID)
if err != nil {
t.Fatalf("pause job: %v", err)
}
archived := putLifecycleTestJob(t, store, "archived", JobLifecycleReady)
if err := store.DeleteJob(context.Background(), archived.ID); err != nil {
t.Fatalf("archive job: %v", err)
}
archived, err = store.GetJob(context.Background(), archived.ID)
if err != nil {
t.Fatalf("get archived job: %v", err)
}
for _, test := range []struct {
name string
job JobDefinition
claimable bool
}{
{name: "enabled", job: enabled, claimable: true},
{name: "ready", job: ready, claimable: true},
{name: "draft", job: draft, claimable: false},
{name: "paused", job: paused, claimable: false},
{name: "archived", job: archived, claimable: false},
} {
t.Run(test.name, func(t *testing.T) {
run := createStoredRun(t, store, test.job, RunStatusQueued)
claimed, ok, err := store.ClaimRun(context.Background(), run.ID, time.Now().UnixMilli())
if err != nil {
t.Fatalf("claim run: %v", err)
}
if ok != test.claimable {
t.Fatalf("claimed = %v, want %v", ok, test.claimable)
}
if ok && claimed.OwnerToken == "" {
t.Fatal("claimed run has no fencing token")
}
})
}
}
func TestManagerArchiveCancelsQueuedAndRunningRuns(t *testing.T) {
store := openTestStore(t)
definition := putTestJob(t, store, "queue")
started := make(chan string, 2)
exited := make(chan struct{}, 1)
executor := ExecutorFunc(func(ctx context.Context, request ExecutionRequest, _ RunReporter) (ExecutionOutcome, error) {
started <- request.Run.ID
<-ctx.Done()
exited <- struct{}{}
return ExecutionOutcome{}, context.Cause(ctx)
})
manager := newTestManager(t, store, executor)
running, err := manager.StartRun(context.Background(), definition.ID)
if err != nil {
t.Fatalf("start running run: %v", err)
}
if got := receiveString(t, started); got != running.ID {
t.Fatalf("started run = %s, want %s", got, running.ID)
}
queued, err := manager.StartRun(context.Background(), definition.ID)
if err != nil {
t.Fatalf("start queued run: %v", err)
}
if err := manager.DeleteJob(context.Background(), definition.ID); err != nil {
t.Fatalf("archive job: %v", err)
}
waitRunStatus(t, store, queued.ID, RunStatusCanceled)
waitRunStatus(t, store, running.ID, RunStatusCanceled)
select {
case <-exited:
case <-time.After(time.Second):
t.Fatal("running executor did not observe archive cancellation")
}
select {
case got := <-started:
t.Fatalf("archived queued run unexpectedly executed: %s", got)
case <-time.After(75 * time.Millisecond):
}
archived, err := store.GetJob(context.Background(), definition.ID)
if err != nil || archived.Lifecycle != JobLifecycleArchived || archived.Enabled {
t.Fatalf("archived definition = %#v, err=%v", archived, err)
}
}
func TestManagerPauseAndPausedPutCancelActiveRuns(t *testing.T) {
store := openTestStore(t)
first := putTestJob(t, store, "queue")
second := putTestJob(t, store, "queue")
started := make(chan string, 2)
executor := ExecutorFunc(func(ctx context.Context, request ExecutionRequest, _ RunReporter) (ExecutionOutcome, error) {
started <- request.Run.JobID
<-ctx.Done()
return ExecutionOutcome{}, context.Cause(ctx)
})
manager, err := NewManager(context.Background(), store, executor, ManagerOptions{
SchedulerInterval: time.Hour,
HeartbeatInterval: time.Hour,
MaxConcurrentRuns: 2,
})
if err != nil {
t.Fatalf("new manager: %v", err)
}
t.Cleanup(func() { shutdownTestManager(t, manager) })
firstRun, err := manager.StartRun(context.Background(), first.ID)
if err != nil {
t.Fatalf("start first run: %v", err)
}
secondRun, err := manager.StartRun(context.Background(), second.ID)
if err != nil {
t.Fatalf("start second run: %v", err)
}
seen := map[string]bool{receiveString(t, started): true, receiveString(t, started): true}
if !seen[first.ID] || !seen[second.ID] {
t.Fatalf("started jobs = %#v", seen)
}
paused, err := manager.PauseJob(context.Background(), first.ID)
if err != nil || paused.Lifecycle != JobLifecyclePaused || paused.Enabled {
t.Fatalf("pause job = %#v, err=%v", paused, err)
}
second.Lifecycle = JobLifecyclePaused
second.Enabled = false
pausedByPut, err := manager.PutJob(context.Background(), second)
if err != nil || pausedByPut.Lifecycle != JobLifecyclePaused || pausedByPut.Enabled {
t.Fatalf("put paused job = %#v, err=%v", pausedByPut, err)
}
waitRunStatus(t, store, firstRun.ID, RunStatusCanceled)
waitRunStatus(t, store, secondRun.ID, RunStatusCanceled)
}
func TestManagerEnforcesMaximumConcurrentRunsAcrossJobs(t *testing.T) {
store := openTestStore(t)
jobs := make([]JobDefinition, 0, 5)
for index := 0; index < 5; index++ {
jobs = append(jobs, putTestJob(t, store, "queue"))
}
started := make(chan string, len(jobs))
release := make(chan struct{}, len(jobs))
var active atomic.Int32
var maximum atomic.Int32
executor := ExecutorFunc(func(ctx context.Context, request ExecutionRequest, _ RunReporter) (ExecutionOutcome, error) {
current := active.Add(1)
defer active.Add(-1)
for {
seen := maximum.Load()
if current <= seen || maximum.CompareAndSwap(seen, current) {
break
}
}
started <- request.Run.ID
select {
case <-ctx.Done():
return ExecutionOutcome{}, context.Cause(ctx)
case <-release:
return ExecutionOutcome{}, nil
}
})
manager, err := NewManager(context.Background(), store, executor, ManagerOptions{
SchedulerInterval: time.Hour,
HeartbeatInterval: time.Hour,
MaxConcurrentRuns: 2,
})
if err != nil {
t.Fatalf("new manager: %v", err)
}
t.Cleanup(func() { shutdownTestManager(t, manager) })
runs := make([]RunRecord, 0, len(jobs))
for _, job := range jobs {
run, err := manager.StartRun(context.Background(), job.ID)
if err != nil {
t.Fatalf("start run: %v", err)
}
runs = append(runs, run)
}
_ = receiveString(t, started)
_ = receiveString(t, started)
select {
case runID := <-started:
t.Fatalf("third run exceeded concurrency limit: %s", runID)
case <-time.After(75 * time.Millisecond):
}
for index := 0; index < len(jobs); index++ {
release <- struct{}{}
}
for _, run := range runs {
waitRunStatus(t, store, run.ID, RunStatusSucceeded)
}
if got := maximum.Load(); got != 2 {
t.Fatalf("maximum concurrent runs = %d, want 2", got)
}
}
func TestStoreRunOwnershipFencesStaleExecutorMutations(t *testing.T) {
store := openTestStore(t)
definition := putTestJob(t, store, "queue")
first := createStoredRun(t, store, definition, RunStatusQueued)
first, claimed, err := store.ClaimRun(context.Background(), first.ID, time.Now().UnixMilli())
if err != nil || !claimed || first.OwnerToken == "" {
t.Fatalf("claim first run = %#v, claimed=%v, err=%v", first, claimed, err)
}
if _, err := store.PutCheckpointOwned(context.Background(), testCheckpoint(definition, first, 1), first.OwnerToken); err != nil {
t.Fatalf("save first checkpoint: %v", err)
}
recovered, err := store.InterruptStaleRuns(context.Background(), first.HeartbeatAt+1, first.HeartbeatAt+2)
if err != nil || len(recovered) != 1 || recovered[0].ID != first.ID {
t.Fatalf("recover first run = %#v, err=%v", recovered, err)
}
second := createStoredRun(t, store, definition, RunStatusQueued)
second, claimed, err = store.ClaimRun(context.Background(), second.ID, first.HeartbeatAt+3)
if err != nil || !claimed || second.OwnerToken == "" || second.OwnerToken == first.OwnerToken {
t.Fatalf("claim second run = %#v, claimed=%v, err=%v", second, claimed, err)
}
if _, err := store.PutCheckpointOwned(context.Background(), testCheckpoint(definition, second, 2), second.OwnerToken); err != nil {
t.Fatalf("save second checkpoint: %v", err)
}
if err := store.TouchRun(context.Background(), second.ID, first.HeartbeatAt+4); !errors.Is(err, ErrRunOwnershipLost) {
t.Fatalf("unowned heartbeat bypass error = %v, want ErrRunOwnershipLost", err)
}
if _, err := store.PutCheckpoint(context.Background(), testCheckpoint(definition, second, 3)); !errors.Is(err, ErrRunOwnershipLost) {
t.Fatalf("unowned checkpoint bypass error = %v, want ErrRunOwnershipLost", err)
}
if err := store.DeleteCheckpoint(context.Background(), definition.ID); !errors.Is(err, ErrRunOwnershipLost) {
t.Fatalf("unowned checkpoint delete bypass error = %v, want ErrRunOwnershipLost", err)
}
if _, err := store.CompleteRun(context.Background(), second.ID, RunStatusSucceeded, ExecutionOutcome{}, "unowned success", first.HeartbeatAt+4); !errors.Is(err, ErrRunOwnershipLost) {
t.Fatalf("unowned completion bypass error = %v, want ErrRunOwnershipLost", err)
}
if err := store.TouchRunOwned(context.Background(), first.ID, first.OwnerToken, first.HeartbeatAt+4); !errors.Is(err, ErrRunOwnershipLost) {
t.Fatalf("stale heartbeat error = %v, want ErrRunOwnershipLost", err)
}
if _, err := store.UpdateRunProgressOwned(context.Background(), first.ID, first.OwnerToken, RunProgress{Current: 1, Total: 1}, first.HeartbeatAt+4); !errors.Is(err, ErrRunOwnershipLost) {
t.Fatalf("stale progress error = %v, want ErrRunOwnershipLost", err)
}
if _, err := store.PutCheckpointOwned(context.Background(), testCheckpoint(definition, first, 99), first.OwnerToken); !errors.Is(err, ErrRunOwnershipLost) {
t.Fatalf("stale checkpoint error = %v, want ErrRunOwnershipLost", err)
}
if err := store.DeleteCheckpointOwned(context.Background(), definition.ID, first.ID, first.OwnerToken); !errors.Is(err, ErrRunOwnershipLost) {
t.Fatalf("stale checkpoint delete error = %v, want ErrRunOwnershipLost", err)
}
if _, err := store.CompleteRunOwned(context.Background(), first.ID, first.OwnerToken, RunStatusSucceeded, ExecutionOutcome{}, "stale success", first.HeartbeatAt+4); !errors.Is(err, ErrRunOwnershipLost) {
t.Fatalf("stale completion error = %v, want ErrRunOwnershipLost", err)
}
checkpoint, err := store.GetCheckpoint(context.Background(), definition.ID)
if err != nil || checkpoint.RunID != second.ID || checkpoint.BatchSequence != 2 {
t.Fatalf("checkpoint after stale writes = %#v, err=%v", checkpoint, err)
}
if _, err := store.CompleteRunOwned(context.Background(), second.ID, second.OwnerToken, RunStatusSucceeded, ExecutionOutcome{}, "", first.HeartbeatAt+5); err != nil {
t.Fatalf("complete second run: %v", err)
}
}
func TestManagerHeartbeatCancelsExecutorAfterOwnershipLoss(t *testing.T) {
store := openTestStore(t)
definition := putTestJob(t, store, "queue")
started := make(chan string, 1)
exited := make(chan error, 1)
executor := ExecutorFunc(func(ctx context.Context, request ExecutionRequest, _ RunReporter) (ExecutionOutcome, error) {
started <- request.Run.ID
<-ctx.Done()
exited <- context.Cause(ctx)
return ExecutionOutcome{}, nil
})
manager, err := NewManager(context.Background(), store, executor, ManagerOptions{
SchedulerInterval: time.Hour,
HeartbeatInterval: 10 * time.Millisecond,
RecoveryStaleAfter: time.Hour,
})
if err != nil {
t.Fatalf("new manager: %v", err)
}
t.Cleanup(func() { shutdownTestManager(t, manager) })
run, err := manager.StartRun(context.Background(), definition.ID)
if err != nil {
t.Fatalf("start run: %v", err)
}
if got := receiveString(t, started); got != run.ID {
t.Fatalf("started run = %s, want %s", got, run.ID)
}
if _, err := store.db.ExecContext(context.Background(), `UPDATE data_sync_runs SET owner_token = 'replacement-owner' WHERE id = ?`, run.ID); err != nil {
t.Fatalf("replace run owner: %v", err)
}
select {
case cause := <-exited:
if !errors.Is(cause, ErrRunOwnershipLost) {
t.Fatalf("executor cancellation cause = %v, want ErrRunOwnershipLost", cause)
}
case <-time.After(time.Second):
t.Fatal("executor was not canceled after ownership loss")
}
deadline := time.Now().Add(time.Second)
for time.Now().Before(deadline) {
manager.mu.Lock()
_, active := manager.active[run.ID]
manager.mu.Unlock()
if !active {
break
}
time.Sleep(time.Millisecond)
}
persisted, err := store.GetRun(context.Background(), run.ID)
if err != nil {
t.Fatalf("get fenced run: %v", err)
}
if persisted.Status != RunStatusRunning || persisted.OwnerToken != "replacement-owner" {
t.Fatalf("stale executor overwrote fenced run: %#v", persisted)
}
}
func TestStandbyLeaseHolderPeriodicallyRecoversStaleRuns(t *testing.T) {
store := openTestStore(t)
definition := putTestJob(t, store, "queue")
run := createStoredRun(t, store, definition, RunStatusQueued)
run, claimed, err := store.ClaimRun(context.Background(), run.ID, time.Now().UnixMilli())
if err != nil || !claimed {
t.Fatalf("claim run: claimed=%v, err=%v", claimed, err)
}
now := time.Now()
if acquired, err := store.AcquireSchedulerLease(context.Background(), "data-sync-scheduler", "primary", now, 100*time.Millisecond); err != nil || !acquired {
t.Fatalf("prime scheduler lease: acquired=%v, err=%v", acquired, err)
}
executor := ExecutorFunc(func(context.Context, ExecutionRequest, RunReporter) (ExecutionOutcome, error) {
t.Fatal("recovered direct run must not execute")
return ExecutionOutcome{}, nil
})
primary, err := NewManager(context.Background(), store, executor, ManagerOptions{
SchedulerInterval: 10 * time.Millisecond,
LeaseTTL: 100 * time.Millisecond,
HeartbeatInterval: time.Hour,
RecoveryStaleAfter: 120 * time.Millisecond,
RecoveryInterval: 20 * time.Millisecond,
LeaseOwner: "primary",
})
if err != nil {
t.Fatalf("new primary manager: %v", err)
}
standby, err := NewManager(context.Background(), store, executor, ManagerOptions{
SchedulerInterval: 10 * time.Millisecond,
LeaseTTL: 100 * time.Millisecond,
HeartbeatInterval: time.Hour,
RecoveryStaleAfter: 120 * time.Millisecond,
RecoveryInterval: 20 * time.Millisecond,
LeaseOwner: "standby",
})
if err != nil {
shutdownTestManager(t, primary)
t.Fatalf("new standby manager: %v", err)
}
t.Cleanup(func() { shutdownTestManager(t, standby) })
shutdownTestManager(t, primary)
recovered := waitRunStatus(t, store, run.ID, RunStatusInterrupted)
if !recovered.Resumable {
t.Fatalf("recovered run is not resumable: %#v", recovered)
}
}
func TestManagerRestartFinalizesStaleCancellingWithoutAutoResume(t *testing.T) {
store := openTestStore(t)
definition := putTestJob(t, store, "forbid")
definition.ResumePolicy = "auto"
definition, err := store.PutJob(context.Background(), definition)
if err != nil {
t.Fatalf("enable auto resume: %v", err)
}
snapshot, err := json.Marshal(definition)
if err != nil {
t.Fatalf("marshal definition: %v", err)
}
staleAt := time.Now().Add(-time.Hour).UnixMilli()
stale, err := store.CreateRun(context.Background(), RunRecord{
JobID: definition.ID,
JobRevision: definition.Revision,
Status: RunStatusRunning,
StartedAt: staleAt,
HeartbeatAt: staleAt,
DefinitionSnapshot: snapshot,
})
if err != nil {
t.Fatalf("create stale run: %v", err)
}
if _, err := store.PutCheckpoint(context.Background(), testCheckpoint(definition, stale, 1)); err != nil {
t.Fatalf("put stale checkpoint: %v", err)
}
if _, err := store.RequestCancelRun(context.Background(), stale.ID, staleAt+1); err != nil {
t.Fatalf("request stale cancellation: %v", err)
}
var executions atomic.Int32
manager, err := NewManager(context.Background(), store, ExecutorFunc(func(context.Context, ExecutionRequest, RunReporter) (ExecutionOutcome, error) {
executions.Add(1)
return ExecutionOutcome{}, nil
}), ManagerOptions{
SchedulerInterval: time.Hour,
HeartbeatInterval: time.Hour,
RecoveryStaleAfter: time.Second,
})
if err != nil {
t.Fatalf("restart manager: %v", err)
}
t.Cleanup(func() { shutdownTestManager(t, manager) })
recovered, err := store.GetRun(context.Background(), stale.ID)
if err != nil {
t.Fatalf("get recovered cancellation: %v", err)
}
if recovered.Status != RunStatusCanceled || recovered.Resumable || recovered.OwnerToken != "" {
t.Fatalf("recovered cancellation = %#v", recovered)
}
events, err := store.ListRunEvents(context.Background(), stale.ID, 0, 10)
if err != nil {
t.Fatalf("list recovery events: %v", err)
}
if len(events) != 1 || events[0].Type != RunEventCanceled {
t.Fatalf("recovery events = %#v", events)
}
time.Sleep(50 * time.Millisecond)
runs, err := store.ListRuns(context.Background(), definition.ID, 10)
if err != nil {
t.Fatalf("list runs after recovery: %v", err)
}
if len(runs) != 1 || executions.Load() != 0 {
t.Fatalf("cancellation resumed after restart: runs=%#v executions=%d", runs, executions.Load())
}
}
func TestManagerResumeAcceptsCheckpointFromAncestorRun(t *testing.T) {
store := openTestStore(t)
definition := putTestJob(t, store, "forbid")
requests := make(chan ExecutionRequest, 3)
var calls atomic.Int32
executor := ExecutorFunc(func(_ context.Context, request ExecutionRequest, reporter RunReporter) (ExecutionOutcome, error) {
requests <- request
switch calls.Add(1) {
case 1:
if err := reporter.SaveCheckpoint(Checkpoint{
Kind: "watermark", Table: "orders", Phase: "copy", CursorType: "primary_key", Cursor: json.RawMessage(`{"id":1}`),
}); err != nil {
return ExecutionOutcome{}, err
}
return ExecutionOutcome{Resumable: true}, errors.New("first failure")
case 2:
return ExecutionOutcome{Resumable: true}, errors.New("resume failed before checkpoint")
default:
return ExecutionOutcome{}, nil
}
})
manager := newTestManager(t, store, executor)
first, err := manager.StartRun(context.Background(), definition.ID)
if err != nil {
t.Fatalf("start first run: %v", err)
}
firstRequest := receiveRequest(t, requests)
if firstRequest.Run.ID != first.ID {
t.Fatalf("first request run = %s, want %s", firstRequest.Run.ID, first.ID)
}
waitRunStatus(t, store, first.ID, RunStatusFailed)
second, err := manager.ResumeRun(context.Background(), first.ID)
if err != nil {
t.Fatalf("resume first run: %v", err)
}
secondRequest := receiveRequest(t, requests)
if secondRequest.Run.ID != second.ID || secondRequest.Checkpoint == nil || secondRequest.Checkpoint.RunID != first.ID {
t.Fatalf("second request = %#v", secondRequest)
}
waitRunStatus(t, store, second.ID, RunStatusFailed)
third, err := manager.ResumeRun(context.Background(), second.ID)
if err != nil {
t.Fatalf("resume second run from ancestor checkpoint: %v", err)
}
thirdRequest := receiveRequest(t, requests)
if thirdRequest.Run.ID != third.ID || thirdRequest.Checkpoint == nil || thirdRequest.Checkpoint.RunID != first.ID {
t.Fatalf("third request = %#v", thirdRequest)
}
waitRunStatus(t, store, third.ID, RunStatusSucceeded)
}
func TestManagerRejectsResumeAndRetryForInsertOnlyRuns(t *testing.T) {
store := openTestStore(t)
definition := putTestJob(t, store, "forbid")
definition.Options.SyncMode = "insert_only"
definition, err := store.PutJob(context.Background(), definition)
if err != nil {
t.Fatalf("enable insert-only mode: %v", err)
}
executor := ExecutorFunc(func(_ context.Context, _ ExecutionRequest, reporter RunReporter) (ExecutionOutcome, error) {
if err := reporter.SaveCheckpoint(Checkpoint{
Kind: "watermark", Table: "orders", Phase: "copy", CursorType: "primary_key", Cursor: json.RawMessage(`{"id":1}`),
}); err != nil {
return ExecutionOutcome{}, err
}
return ExecutionOutcome{Resumable: true}, errors.New("partial insert-only failure")
})
manager := newTestManager(t, store, executor)
run, err := manager.StartRun(context.Background(), definition.ID)
if err != nil {
t.Fatalf("start insert-only run: %v", err)
}
waitRunStatus(t, store, run.ID, RunStatusFailed)
if _, err := manager.ResumeRun(context.Background(), run.ID); !errors.Is(err, ErrRunNotResumable) {
t.Fatalf("insert-only resume error = %v, want ErrRunNotResumable", err)
}
if _, err := manager.RetryRun(context.Background(), run.ID); !errors.Is(err, ErrRunNotRetryable) {
t.Fatalf("insert-only retry error = %v, want ErrRunNotRetryable", err)
}
}
func TestStoreMigratesRunOwnershipColumnFromVersionTwo(t *testing.T) {
path := t.TempDir() + "/sync-jobs.db"
store, err := Open(path)
if err != nil {
t.Fatalf("create current store: %v", err)
}
if err := store.Close(); err != nil {
t.Fatalf("close current store: %v", err)
}
database, err := sql.Open("sqlite", sqliteDSN(path))
if err != nil {
t.Fatalf("open raw store: %v", err)
}
if _, err := database.Exec(`ALTER TABLE data_sync_runs DROP COLUMN owner_token`); err != nil {
_ = database.Close()
t.Fatalf("remove ownership column: %v", err)
}
if _, err := database.Exec(`PRAGMA user_version=2`); err != nil {
_ = database.Close()
t.Fatalf("downgrade schema marker: %v", err)
}
if err := database.Close(); err != nil {
t.Fatalf("close raw store: %v", err)
}
migrated, err := Open(path)
if err != nil {
t.Fatalf("migrate version two store: %v", err)
}
t.Cleanup(func() { _ = migrated.Close() })
hasColumn, err := sqliteTableHasColumn(context.Background(), migrated.db, "data_sync_runs", "owner_token")
if err != nil || !hasColumn {
t.Fatalf("owner_token column present = %v, err=%v", hasColumn, err)
}
}
func TestStoreMigratesErrorRowRetryLeaseFromVersionThree(t *testing.T) {
path := t.TempDir() + "/sync-jobs.db"
store, err := Open(path)
if err != nil {
t.Fatalf("create current store: %v", err)
}
if err := store.Close(); err != nil {
t.Fatalf("close current store: %v", err)
}
database, err := sql.Open("sqlite", sqliteDSN(path))
if err != nil {
t.Fatalf("open raw store: %v", err)
}
for _, statement := range []string{
`DROP INDEX idx_data_sync_error_rows_retry`,
`ALTER TABLE data_sync_error_rows DROP COLUMN retry_owner`,
`ALTER TABLE data_sync_error_rows DROP COLUMN retry_lease_expires_at`,
`PRAGMA user_version=3`,
} {
if _, err := database.Exec(statement); err != nil {
_ = database.Close()
t.Fatalf("prepare version three store (%s): %v", statement, err)
}
}
if err := database.Close(); err != nil {
t.Fatalf("close raw store: %v", err)
}
migrated, err := Open(path)
if err != nil {
t.Fatalf("migrate version three store: %v", err)
}
t.Cleanup(func() { _ = migrated.Close() })
for _, column := range []string{"retry_owner", "retry_lease_expires_at"} {
hasColumn, err := sqliteTableHasColumn(context.Background(), migrated.db, "data_sync_error_rows", column)
if err != nil || !hasColumn {
t.Fatalf("%s column present = %v, err=%v", column, hasColumn, err)
}
}
}
func TestManagerRunCreationRollsBackWhenQueuedEventCannotPersist(t *testing.T) {
store := openTestStore(t)
definition := putTestJob(t, store, "queue")
if _, err := store.db.ExecContext(context.Background(), `CREATE TRIGGER fail_queued_run_event
BEFORE INSERT ON data_sync_run_events WHEN NEW.event_type = 'queued'
BEGIN SELECT RAISE(ABORT, 'injected queued event failure'); END`); err != nil {
t.Fatalf("create event failure trigger: %v", err)
}
manager := newTestManager(t, store, ExecutorFunc(func(context.Context, ExecutionRequest, RunReporter) (ExecutionOutcome, error) {
t.Fatal("run without queued event must not execute")
return ExecutionOutcome{}, nil
}))
if _, err := manager.StartRun(context.Background(), definition.ID); err == nil {
t.Fatal("start run succeeded despite queued event failure")
}
runs, err := store.ListRuns(context.Background(), definition.ID, 10)
if err != nil {
t.Fatalf("list runs: %v", err)
}
if len(runs) != 0 {
t.Fatalf("run persisted without queued event: %#v", runs)
}
}
func putLifecycleTestJob(t *testing.T, store *Store, name string, lifecycle JobLifecycle) JobDefinition {
t.Helper()
definition, err := store.PutJob(context.Background(), JobDefinition{
Name: name, Lifecycle: lifecycle, Kind: JobKindReconcile, IncrementalMode: IncrementalSnapshot,
Source: EndpointRef{ConnectionID: "source-" + name}, Target: EndpointRef{ConnectionID: "target-" + name},
Mappings: []TableMapping{{SourceTable: "orders", TargetTable: "orders", Enabled: true}},
})
if err != nil {
t.Fatalf("put %s job: %v", name, err)
}
return definition
}
func createStoredRun(t *testing.T, store *Store, definition JobDefinition, status RunStatus) RunRecord {
t.Helper()
snapshot, err := json.Marshal(definition)
if err != nil {
t.Fatalf("marshal run definition: %v", err)
}
run, err := store.CreateRun(context.Background(), RunRecord{
JobID: definition.ID, JobRevision: definition.Revision, Status: status, DefinitionSnapshot: snapshot,
SourceFingerprint: definition.Source.Fingerprint, TargetFingerprint: definition.Target.Fingerprint,
})
if err != nil {
t.Fatalf("create stored run: %v", err)
}
return run
}
func testCheckpoint(definition JobDefinition, run RunRecord, sequence int64) Checkpoint {
return Checkpoint{
Kind: "watermark", JobID: definition.ID, RunID: run.ID, DefinitionRevision: definition.Revision,
Table: "orders", Phase: "copy", CursorType: "primary_key", Cursor: json.RawMessage(`{"id":1}`), BatchSequence: sequence,
}
}
func shutdownTestManager(t *testing.T, manager *Manager) {
t.Helper()
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
if err := manager.Shutdown(ctx); err != nil {
t.Errorf("shutdown manager: %v", err)
}
}

1865
internal/syncjob/store.go Normal file

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,296 @@
package syncjob
import (
"context"
"encoding/json"
"errors"
"fmt"
"sync"
"testing"
"time"
)
func TestStoreSchedulerLeaseAllowsOnlyTheOwnerUntilExpiry(t *testing.T) {
store := openTestStore(t)
now := time.Now().Truncate(time.Millisecond)
acquired, err := store.AcquireSchedulerLease(context.Background(), "scheduler", "owner-a", now, time.Second)
if err != nil || !acquired {
t.Fatalf("owner-a acquire = %v, %v", acquired, err)
}
acquired, err = store.AcquireSchedulerLease(context.Background(), "scheduler", "owner-b", now.Add(500*time.Millisecond), time.Second)
if err != nil {
t.Fatalf("owner-b acquire before expiry: %v", err)
}
if acquired {
t.Fatal("owner-b acquired a live lease")
}
acquired, err = store.AcquireSchedulerLease(context.Background(), "scheduler", "owner-b", now.Add(time.Second), time.Second)
if err != nil || !acquired {
t.Fatalf("owner-b takeover = %v, %v", acquired, err)
}
if err := store.ReleaseSchedulerLease(context.Background(), "scheduler", "owner-a"); err != nil {
t.Fatalf("old owner release: %v", err)
}
acquired, err = store.AcquireSchedulerLease(context.Background(), "scheduler", "owner-c", now.Add(1500*time.Millisecond), time.Second)
if err != nil {
t.Fatalf("owner-c acquire while owner-b live: %v", err)
}
if acquired {
t.Fatal("old owner release removed the replacement lease")
}
}
func TestStorePersistsConcurrentRunEventsWithContiguousSequences(t *testing.T) {
store := openTestStore(t)
definition := putTestJob(t, store, "queue")
snapshot, err := json.Marshal(definition)
if err != nil {
t.Fatalf("marshal definition: %v", err)
}
run, err := store.CreateRun(context.Background(), RunRecord{
JobID: definition.ID,
JobRevision: definition.Revision,
DefinitionSnapshot: snapshot,
})
if err != nil {
t.Fatalf("create run: %v", err)
}
const count = 32
errorsSeen := make(chan error, count)
var wait sync.WaitGroup
for index := 0; index < count; index++ {
index := index
wait.Add(1)
go func() {
defer wait.Done()
_, appendErr := store.AppendRunEvent(context.Background(), RunEvent{
RunID: run.ID,
Type: RunEventLog,
Message: fmt.Sprintf("event-%d", index),
})
if appendErr != nil {
errorsSeen <- appendErr
}
}()
}
wait.Wait()
close(errorsSeen)
for appendErr := range errorsSeen {
t.Errorf("append event: %v", appendErr)
}
events, err := store.ListRunEvents(context.Background(), run.ID, 0, count)
if err != nil {
t.Fatalf("list events: %v", err)
}
if len(events) != count {
t.Fatalf("event count = %d, want %d", len(events), count)
}
for index, event := range events {
if event.Sequence != int64(index+1) {
t.Fatalf("event sequence at %d = %d, want %d", index, event.Sequence, index+1)
}
}
}
func TestStorePersistsIncompleteDraftButManagerWillNotRunIt(t *testing.T) {
store := openTestStore(t)
draft, err := store.PutJob(context.Background(), JobDefinition{
Name: "unfinished sync",
Lifecycle: JobLifecycleDraft,
})
if err != nil {
t.Fatalf("persist draft: %v", err)
}
if draft.Enabled || draft.NextRunAt != 0 || draft.Lifecycle != JobLifecycleDraft {
t.Fatalf("normalized draft = %#v", draft)
}
manager := newTestManager(t, store, ExecutorFunc(func(context.Context, ExecutionRequest, RunReporter) (ExecutionOutcome, error) {
t.Fatal("draft executor must not run")
return ExecutionOutcome{}, nil
}))
if _, err := manager.StartRun(context.Background(), draft.ID); !errors.Is(err, ErrJobDisabled) {
t.Fatalf("start draft error = %v, want ErrJobDisabled", err)
}
if _, err := store.PutJob(context.Background(), JobDefinition{
Name: "invalid ready job",
Lifecycle: JobLifecycleReady,
}); err == nil {
t.Fatal("ready job without endpoints and mappings was persisted")
}
}
func TestStoreResetCheckpointRejectsActiveRun(t *testing.T) {
store := openTestStore(t)
definition := putTestJob(t, store, "forbid")
snapshot, err := json.Marshal(definition)
if err != nil {
t.Fatalf("marshal definition: %v", err)
}
run, err := store.CreateRun(context.Background(), RunRecord{
JobID: definition.ID,
JobRevision: definition.Revision,
Status: RunStatusRunning,
DefinitionSnapshot: snapshot,
})
if err != nil {
t.Fatalf("create running run: %v", err)
}
if _, err := store.PutCheckpoint(context.Background(), Checkpoint{
Version: 1,
Kind: "watermark",
JobID: definition.ID,
RunID: run.ID,
DefinitionRevision: definition.Revision,
Table: "orders",
Phase: "batch_committed",
CursorType: "watermark_map",
Cursor: json.RawMessage(`{"orders":{"id":42}}`),
}); err != nil {
t.Fatalf("put checkpoint: %v", err)
}
if err := store.ResetCheckpoint(context.Background(), definition.ID); !errors.Is(err, ErrRunAlreadyActive) {
t.Fatalf("reset with active run error = %v, want ErrRunAlreadyActive", err)
}
if _, err := store.CompleteRun(context.Background(), run.ID, RunStatusFailed, ExecutionOutcome{Resumable: true}, "failed", time.Now().UnixMilli()); err != nil {
t.Fatalf("complete run: %v", err)
}
if err := store.ResetCheckpoint(context.Background(), definition.ID); err != nil {
t.Fatalf("reset checkpoint: %v", err)
}
if err := store.ResetCheckpoint(context.Background(), definition.ID); !errors.Is(err, ErrNotFound) {
t.Fatalf("second reset error = %v, want ErrNotFound", err)
}
}
func TestManagerReadsAndDiscardsErrorRowWithOneWayCAS(t *testing.T) {
store := openTestStore(t)
definition := putTestJob(t, store, "queue")
snapshot, err := json.Marshal(definition)
if err != nil {
t.Fatalf("marshal definition: %v", err)
}
run, err := store.CreateRun(context.Background(), RunRecord{
JobID: definition.ID,
JobRevision: definition.Revision,
Status: RunStatusFailed,
DefinitionSnapshot: snapshot,
})
if err != nil {
t.Fatalf("create run: %v", err)
}
row, err := store.AppendErrorRow(context.Background(), ErrorRow{
RunID: run.ID,
JobID: definition.ID,
Error: "duplicate key",
})
if err != nil {
t.Fatalf("append error row: %v", err)
}
manager := newTestManager(t, store, ExecutorFunc(func(context.Context, ExecutionRequest, RunReporter) (ExecutionOutcome, error) {
return ExecutionOutcome{}, nil
}))
read, err := manager.GetErrorRow(context.Background(), row.ID)
if err != nil || read.ID != row.ID || read.Status != ErrorRowPending {
t.Fatalf("get error row = %#v, err=%v", read, err)
}
if err := manager.RecordErrorRowRetryFailure(context.Background(), row.ID); err != nil {
t.Fatalf("record retry failure: %v", err)
}
read, err = manager.GetErrorRow(context.Background(), row.ID)
if err != nil || read.Status != ErrorRowPending || read.Attempts != 1 {
t.Fatalf("pending retried error row = %#v, err=%v", read, err)
}
if err := manager.DiscardErrorRow(context.Background(), row.ID); err != nil {
t.Fatalf("discard error row: %v", err)
}
discarded, err := manager.GetErrorRow(context.Background(), row.ID)
if err != nil || discarded.Status != ErrorRowDiscarded {
t.Fatalf("discarded error row = %#v, err=%v", discarded, err)
}
if err := manager.DiscardErrorRow(context.Background(), row.ID); !errors.Is(err, ErrErrorRowStateConflict) {
t.Fatalf("repeat discard error = %v, want ErrErrorRowStateConflict", err)
}
if err := manager.RecordErrorRowRetryFailure(context.Background(), row.ID); !errors.Is(err, ErrErrorRowStateConflict) {
t.Fatalf("discarded retry failure error = %v, want ErrErrorRowStateConflict", err)
}
if err := store.UpdateErrorRowStatus(context.Background(), row.ID, ErrorRowResolved, true); !errors.Is(err, ErrErrorRowStateConflict) {
t.Fatalf("discarded to resolved error = %v, want ErrErrorRowStateConflict", err)
}
resolvedRow, err := store.AppendErrorRow(context.Background(), ErrorRow{
RunID: run.ID,
JobID: definition.ID,
Error: "timeout",
})
if err != nil {
t.Fatalf("append resolvable error row: %v", err)
}
if err := manager.ResolveErrorRow(context.Background(), resolvedRow.ID, true); err != nil {
t.Fatalf("resolve error row: %v", err)
}
resolved, err := manager.GetErrorRow(context.Background(), resolvedRow.ID)
if err != nil || resolved.Status != ErrorRowResolved || resolved.Attempts != 1 {
t.Fatalf("resolved error row = %#v, err=%v", resolved, err)
}
if err := manager.DiscardErrorRow(context.Background(), resolvedRow.ID); !errors.Is(err, ErrErrorRowStateConflict) {
t.Fatalf("resolved to discarded error = %v, want ErrErrorRowStateConflict", err)
}
}
func TestStoreErrorRowRetryClaimFencesTransitionsAndRecoversExpiredLease(t *testing.T) {
store := openTestStore(t)
definition := putTestJob(t, store, "queue")
run := createStoredRun(t, store, definition, RunStatusFailed)
row, err := store.AppendErrorRow(context.Background(), ErrorRow{
RunID: run.ID,
JobID: definition.ID,
Error: "duplicate key",
})
if err != nil {
t.Fatalf("append error row: %v", err)
}
now := time.Now().UnixMilli()
claimed, err := store.ClaimErrorRowRetry(context.Background(), row.ID, now, time.Second)
if err != nil {
t.Fatalf("claim error row retry: %v", err)
}
if claimed.Status != ErrorRowRetrying || claimed.RetryOwner == "" || claimed.RetryLeaseExpiresAt != now+time.Second.Milliseconds() {
t.Fatalf("claimed error row = %#v", claimed)
}
if _, err := store.ClaimErrorRowRetry(context.Background(), row.ID, now+500, time.Second); !errors.Is(err, ErrErrorRowStateConflict) {
t.Fatalf("concurrent claim error = %v, want ErrErrorRowStateConflict", err)
}
if err := store.UpdateErrorRowStatus(context.Background(), row.ID, ErrorRowDiscarded, false); !errors.Is(err, ErrErrorRowStateConflict) {
t.Fatalf("discard retrying row error = %v, want ErrErrorRowStateConflict", err)
}
if err := store.ResolveErrorRowRetry(context.Background(), row.ID, "wrong-owner", now+600); !errors.Is(err, ErrErrorRowRetryOwnershipLost) {
t.Fatalf("wrong-owner resolution error = %v, want ErrErrorRowRetryOwnershipLost", err)
}
if err := store.RenewErrorRowRetry(context.Background(), row.ID, claimed.RetryOwner, now+500, time.Second); err != nil {
t.Fatalf("renew retry claim: %v", err)
}
if recovered, err := store.RecoverExpiredErrorRowRetries(context.Background(), now+time.Second.Milliseconds()+1); err != nil || recovered != 0 {
t.Fatalf("recover live renewed retry claims = %d, err=%v", recovered, err)
}
recovered, err := store.RecoverExpiredErrorRowRetries(context.Background(), now+1501)
if err != nil || recovered != 1 {
t.Fatalf("recover expired retry claims = %d, err=%v", recovered, err)
}
pending, err := store.GetErrorRow(context.Background(), row.ID)
if err != nil || pending.Status != ErrorRowPending || pending.Attempts != 1 || pending.RetryOwner != "" || pending.RetryLeaseExpiresAt != 0 {
t.Fatalf("recovered retry row = %#v, err=%v", pending, err)
}
reclaimed, err := store.ClaimErrorRowRetry(context.Background(), row.ID, now+2000, time.Second)
if err != nil {
t.Fatalf("reclaim recovered error row: %v", err)
}
if err := store.ResolveErrorRowRetry(context.Background(), row.ID, reclaimed.RetryOwner, now+2100); err != nil {
t.Fatalf("resolve reclaimed error row: %v", err)
}
resolved, err := store.GetErrorRow(context.Background(), row.ID)
if err != nil || resolved.Status != ErrorRowResolved || resolved.Attempts != 2 || resolved.RetryOwner != "" || resolved.RetryLeaseExpiresAt != 0 {
t.Fatalf("resolved retried row = %#v, err=%v", resolved, err)
}
}

View File

@@ -0,0 +1,593 @@
package syncjob
import (
"bytes"
"encoding/json"
"errors"
"fmt"
"strconv"
"strings"
"time"
)
const (
defaultBatchSize = 1000
maxBatchSize = 10000
maxRetries = 10
minScheduleInterval = 10 * time.Second
continuousRunPoll = 5 * time.Second
)
var supportedTransforms = map[string]struct{}{
"": {},
"identity": {},
"string": {},
"int64": {},
"float64": {},
"bool": {},
"timestamp": {},
"date": {},
"json": {},
"lower": {},
"upper": {},
"trim": {},
"constant": {},
"coalesce": {},
}
func NormalizeDefinition(input JobDefinition) JobDefinition {
definition := input
if definition.Version == 0 {
definition.Version = CurrentDefinitionVersion
}
definition.ID = strings.TrimSpace(definition.ID)
definition.Name = strings.TrimSpace(definition.Name)
definition.Description = strings.TrimSpace(definition.Description)
if definition.Lifecycle == "" {
switch {
case definition.ArchivedAt > 0:
definition.Lifecycle = JobLifecycleArchived
case definition.Enabled:
definition.Lifecycle = JobLifecycleEnabled
default:
definition.Lifecycle = JobLifecycleReady
}
}
definition.Enabled = definition.Lifecycle == JobLifecycleEnabled
definition.Source = normalizeEndpoint(definition.Source)
definition.Target = normalizeEndpoint(definition.Target)
if definition.Kind == "" {
definition.Kind = JobKindReconcile
}
if definition.IncrementalMode == "" {
definition.IncrementalMode = IncrementalSnapshot
}
definition.SourceQuery = strings.TrimSpace(definition.SourceQuery)
if definition.CDC != nil {
definition.CDC.Adapter = strings.ToLower(strings.TrimSpace(definition.CDC.Adapter))
definition.CDC.StartPosition = strings.ToLower(strings.TrimSpace(definition.CDC.StartPosition))
definition.CDC.SlotName = strings.TrimSpace(definition.CDC.SlotName)
definition.CDC.PublicationName = strings.TrimSpace(definition.CDC.PublicationName)
}
if definition.Approval != nil {
definition.Approval.DefinitionHash = strings.TrimSpace(definition.Approval.DefinitionHash)
definition.Approval.TargetFingerprint = strings.TrimSpace(definition.Approval.TargetFingerprint)
definition.Approval.ApprovedByRuntime = strings.TrimSpace(definition.Approval.ApprovedByRuntime)
}
if definition.Schedule.Kind == "" {
definition.Schedule.Kind = ScheduleManual
}
if definition.Schedule.MisfirePolicy == "" {
definition.Schedule.MisfirePolicy = "skip"
}
definition.Schedule.CronExpression = strings.TrimSpace(definition.Schedule.CronExpression)
definition.Schedule.Timezone = strings.TrimSpace(definition.Schedule.Timezone)
if definition.Schedule.Timezone == "" {
definition.Schedule.Timezone = "Local"
}
if definition.ConcurrencyPolicy == "" {
definition.ConcurrencyPolicy = "forbid"
}
if definition.ResumePolicy == "" {
definition.ResumePolicy = "manual"
}
if definition.Options.Content == "" {
definition.Options.Content = "data"
}
if definition.Options.SyncMode == "" {
definition.Options.SyncMode = "insert_update"
}
if definition.Options.TargetTableStrategy == "" {
definition.Options.TargetTableStrategy = "existing_only"
}
if definition.Options.BatchSize == 0 {
definition.Options.BatchSize = defaultBatchSize
}
if definition.Options.ErrorPolicy == "" {
definition.Options.ErrorPolicy = ErrorPolicyStop
}
if definition.Options.RetryBackoffMillis == 0 {
definition.Options.RetryBackoffMillis = 500
}
for index := range definition.Mappings {
mapping := &definition.Mappings[index]
mapping.SourceSchema = strings.TrimSpace(mapping.SourceSchema)
mapping.SourceTable = strings.TrimSpace(mapping.SourceTable)
mapping.TargetSchema = strings.TrimSpace(mapping.TargetSchema)
mapping.TargetTable = strings.TrimSpace(mapping.TargetTable)
mapping.TargetTableStrategy = strings.ToLower(strings.TrimSpace(mapping.TargetTableStrategy))
mapping.Filter = strings.TrimSpace(mapping.Filter)
mapping.KeyColumns = normalizeUniqueStrings(mapping.KeyColumns)
for columnIndex := range mapping.Columns {
column := &mapping.Columns[columnIndex]
column.Source = strings.TrimSpace(column.Source)
column.Target = strings.TrimSpace(column.Target)
column.Transform.Kind = strings.ToLower(strings.TrimSpace(column.Transform.Kind))
}
if mapping.Watermark != nil {
mapping.Watermark.Column = strings.TrimSpace(mapping.Watermark.Column)
mapping.Watermark.TieBreakerColumns = normalizeUniqueStrings(mapping.Watermark.TieBreakerColumns)
}
}
return definition
}
func ValidateDefinition(input JobDefinition) error {
definition := NormalizeDefinition(input)
if err := validatePersistableDefinitionEnums(definition); err != nil {
return err
}
if definition.Source.ConnectionID == "" {
return errors.New("source saved connection is required")
}
if definition.Target.ConnectionID == "" {
return errors.New("target saved connection is required")
}
if definition.Approval != nil {
if definition.Approval.DefinitionHash == "" || definition.Approval.TargetFingerprint == "" || definition.Approval.ApprovedAt <= 0 || definition.Approval.ApprovedByRuntime == "" {
return errors.New("execution approval requires definitionHash, targetFingerprint, approvedAt, and approvedByRuntime")
}
}
switch definition.Kind {
case JobKindMigration, JobKindReconcile, JobKindQuerySink, JobKindCompare:
default:
return fmt.Errorf("unsupported data sync job kind %q", definition.Kind)
}
switch definition.IncrementalMode {
case IncrementalSnapshot, IncrementalWatermark, IncrementalCDC:
default:
return fmt.Errorf("unsupported incremental mode %q", definition.IncrementalMode)
}
if definition.Kind == JobKindQuerySink && definition.SourceQuery == "" {
return errors.New("query sink jobs require sourceQuery")
}
if definition.Kind != JobKindQuerySink && definition.SourceQuery != "" {
return errors.New("sourceQuery is only supported by query sink jobs")
}
if len(definition.Mappings) == 0 {
return errors.New("at least one table mapping is required")
}
if definition.Kind == JobKindQuerySink && len(definition.Mappings) != 1 {
return errors.New("query sink jobs require exactly one target mapping")
}
if (definition.Kind == JobKindQuerySink || definition.Kind == JobKindCompare) && definition.IncrementalMode != IncrementalSnapshot {
return fmt.Errorf("%s jobs only support snapshot execution", definition.Kind)
}
seenTargets := make(map[string]struct{}, len(definition.Mappings))
enabledMappings := 0
for index, mapping := range definition.Mappings {
if !mapping.Enabled {
continue
}
enabledMappings++
if mapping.TargetTable == "" || (definition.Kind != JobKindQuerySink && mapping.SourceTable == "") {
return fmt.Errorf("table mapping %d requires a targetTable and a sourceTable unless this is a query sink", index+1)
}
switch mapping.TargetTableStrategy {
case "", "existing_only", "auto_create_if_missing", "smart":
default:
return fmt.Errorf("table mapping %s has unsupported targetTableStrategy %q", mapping.SourceTable, mapping.TargetTableStrategy)
}
targetKey := strings.ToLower(mapping.TargetSchema + "\x00" + mapping.TargetTable)
if _, exists := seenTargets[targetKey]; exists {
return fmt.Errorf("duplicate target table mapping %s", mapping.TargetTable)
}
seenTargets[targetKey] = struct{}{}
if err := validateColumnMappings(mapping); err != nil {
return fmt.Errorf("table mapping %s: %w", mapping.SourceTable, err)
}
if definition.IncrementalMode == IncrementalWatermark {
if mapping.Watermark == nil || strings.TrimSpace(mapping.Watermark.Column) == "" {
return fmt.Errorf("table mapping %s requires a watermark column", mapping.SourceTable)
}
}
if definition.IncrementalMode == IncrementalCDC && len(mapping.KeyColumns) == 0 {
return fmt.Errorf("table mapping %s requires stable keyColumns for CDC", mapping.SourceTable)
}
}
if enabledMappings == 0 {
return errors.New("at least one table mapping must be enabled")
}
if definition.IncrementalMode == IncrementalCDC {
if definition.CDC == nil || strings.TrimSpace(definition.CDC.Adapter) == "" {
return errors.New("CDC jobs require an explicit adapter")
}
switch definition.CDC.StartPosition {
case "", "checkpoint", "latest", "earliest":
default:
return fmt.Errorf("unsupported CDC start position %q", definition.CDC.StartPosition)
}
if definition.Options.TargetTableStrategy != "existing_only" {
return errors.New("CDC jobs require existing target tables")
}
}
if definition.Options.BatchSize < 1 || definition.Options.BatchSize > maxBatchSize {
return fmt.Errorf("batchSize must be between 1 and %d", maxBatchSize)
}
if definition.Options.MaxRetries < 0 || definition.Options.MaxRetries > maxRetries {
return fmt.Errorf("maxRetries must be between 0 and %d", maxRetries)
}
if definition.Options.RetryBackoffMillis < 0 || definition.Options.RetryBackoffMillis > int((5*time.Minute)/time.Millisecond) {
return errors.New("retryBackoffMillis must be between 0 and 300000")
}
switch definition.Options.ErrorPolicy {
case ErrorPolicyStop, ErrorPolicySkipRow:
default:
return fmt.Errorf("unsupported error policy %q", definition.Options.ErrorPolicy)
}
switch definition.Schedule.Kind {
case ScheduleManual:
case ScheduleOnce:
if definition.Schedule.RunAt <= 0 {
return errors.New("one-time schedules require runAt")
}
case ScheduleInterval:
if time.Duration(definition.Schedule.IntervalSeconds)*time.Second < minScheduleInterval {
return fmt.Errorf("scheduled interval must be at least %s", minScheduleInterval)
}
case ScheduleCron:
if _, err := parseCronSchedule(definition.Schedule.CronExpression, definition.Schedule.Timezone); err != nil {
return err
}
case ScheduleContinuous:
if definition.IncrementalMode != IncrementalCDC {
return errors.New("continuous trigger requires CDC incremental mode")
}
if definition.ConcurrencyPolicy != "forbid" {
return errors.New("continuous trigger requires forbid concurrency policy")
}
default:
return fmt.Errorf("unsupported schedule kind %q", definition.Schedule.Kind)
}
switch definition.Schedule.MisfirePolicy {
case "skip", "run_once", "catch_up":
default:
return fmt.Errorf("unsupported misfire policy %q", definition.Schedule.MisfirePolicy)
}
switch definition.ConcurrencyPolicy {
case "forbid", "queue":
default:
return fmt.Errorf("unsupported concurrency policy %q", definition.ConcurrencyPolicy)
}
switch definition.ResumePolicy {
case "never", "manual", "auto":
default:
return fmt.Errorf("unsupported resume policy %q", definition.ResumePolicy)
}
return nil
}
func ValidatePersistableDefinition(input JobDefinition) error {
definition := NormalizeDefinition(input)
if err := validatePersistableDefinitionEnums(definition); err != nil {
return err
}
if definition.Lifecycle == JobLifecycleDraft || definition.Lifecycle == JobLifecycleArchived {
return nil
}
return ValidateDefinition(definition)
}
func validatePersistableDefinitionEnums(definition JobDefinition) error {
if definition.Version != CurrentDefinitionVersion {
return fmt.Errorf("unsupported data sync job definition version %d", definition.Version)
}
if definition.Name == "" {
return errors.New("data sync job name is required")
}
switch definition.Lifecycle {
case JobLifecycleDraft, JobLifecycleReady, JobLifecycleEnabled, JobLifecyclePaused, JobLifecycleArchived:
default:
return fmt.Errorf("unsupported data sync job lifecycle %q", definition.Lifecycle)
}
switch definition.Kind {
case JobKindMigration, JobKindReconcile, JobKindQuerySink, JobKindCompare:
default:
return fmt.Errorf("unsupported data sync job kind %q", definition.Kind)
}
switch definition.IncrementalMode {
case IncrementalSnapshot, IncrementalWatermark, IncrementalCDC:
default:
return fmt.Errorf("unsupported incremental mode %q", definition.IncrementalMode)
}
switch definition.Schedule.Kind {
case ScheduleManual, ScheduleOnce, ScheduleInterval, ScheduleCron, ScheduleContinuous:
default:
return fmt.Errorf("unsupported schedule kind %q", definition.Schedule.Kind)
}
switch definition.Schedule.MisfirePolicy {
case "skip", "run_once", "catch_up":
default:
return fmt.Errorf("unsupported misfire policy %q", definition.Schedule.MisfirePolicy)
}
switch definition.ConcurrencyPolicy {
case "forbid", "queue":
default:
return fmt.Errorf("unsupported concurrency policy %q", definition.ConcurrencyPolicy)
}
switch definition.ResumePolicy {
case "never", "manual", "auto":
default:
return fmt.Errorf("unsupported resume policy %q", definition.ResumePolicy)
}
switch definition.Options.ErrorPolicy {
case ErrorPolicyStop, ErrorPolicySkipRow:
default:
return fmt.Errorf("unsupported error policy %q", definition.Options.ErrorPolicy)
}
return nil
}
func validateColumnMappings(mapping TableMapping) error {
seenTargets := make(map[string]struct{}, len(mapping.Columns))
for index, column := range mapping.Columns {
if column.Target == "" {
return fmt.Errorf("column mapping %d requires target", index+1)
}
if column.Source == "" && column.Transform.Kind != "constant" && len(bytes.TrimSpace(column.DefaultValue)) == 0 {
return fmt.Errorf("column mapping %s requires source, constant transform, or defaultValue", column.Target)
}
targetKey := strings.ToLower(column.Target)
if _, exists := seenTargets[targetKey]; exists {
return fmt.Errorf("duplicate target column %s", column.Target)
}
seenTargets[targetKey] = struct{}{}
if _, ok := supportedTransforms[column.Transform.Kind]; !ok {
return fmt.Errorf("unsupported transform %q", column.Transform.Kind)
}
if !validJSONOrEmpty(column.Transform.Argument) {
return fmt.Errorf("transform argument for %s is not valid JSON", column.Target)
}
if !validJSONOrEmpty(column.DefaultValue) {
return fmt.Errorf("default value for %s is not valid JSON", column.Target)
}
}
return nil
}
func normalizeEndpoint(endpoint EndpointRef) EndpointRef {
endpoint.ConnectionID = strings.TrimSpace(endpoint.ConnectionID)
endpoint.ConnectionType = strings.ToLower(strings.TrimSpace(endpoint.ConnectionType))
endpoint.ConnectionName = strings.TrimSpace(endpoint.ConnectionName)
endpoint.Database = strings.TrimSpace(endpoint.Database)
endpoint.Schema = strings.TrimSpace(endpoint.Schema)
endpoint.Fingerprint = strings.TrimSpace(endpoint.Fingerprint)
return endpoint
}
func normalizeUniqueStrings(values []string) []string {
result := make([]string, 0, len(values))
seen := make(map[string]struct{}, len(values))
for _, value := range values {
trimmed := strings.TrimSpace(value)
if trimmed == "" {
continue
}
key := strings.ToLower(trimmed)
if _, exists := seen[key]; exists {
continue
}
seen[key] = struct{}{}
result = append(result, trimmed)
}
return result
}
func validJSONOrEmpty(raw json.RawMessage) bool {
trimmed := bytes.TrimSpace(raw)
return len(trimmed) == 0 || json.Valid(trimmed)
}
func NextRunAt(definition JobDefinition, after time.Time) int64 {
definition = NormalizeDefinition(definition)
if !definition.Enabled {
return 0
}
switch definition.Schedule.Kind {
case ScheduleOnce:
if definition.Schedule.RunAt > after.UnixMilli() {
return definition.Schedule.RunAt
}
return 0
case ScheduleCron:
next, err := nextCronTime(definition.Schedule.CronExpression, definition.Schedule.Timezone, after)
if err != nil {
return 0
}
return next.UnixMilli()
case ScheduleInterval:
if definition.Schedule.IntervalSeconds <= 0 {
return 0
}
case ScheduleContinuous:
// Continuous jobs are reconciled by the leased scheduler. While a stream
// is active the forbid policy suppresses duplicates; after EOF/failure the
// next poll starts a fresh run from the durable CDC checkpoint.
return after.Add(continuousRunPoll).UnixMilli()
default:
return 0
}
interval := time.Duration(definition.Schedule.IntervalSeconds) * time.Second
anchorMillis := definition.Schedule.AnchorAt
if anchorMillis <= 0 {
return after.Add(interval).UnixMilli()
}
anchor := time.UnixMilli(anchorMillis)
if after.Before(anchor) {
return anchor.UnixMilli()
}
steps := after.Sub(anchor)/interval + 1
return anchor.Add(steps * interval).UnixMilli()
}
type cronSchedule struct {
minutes map[int]struct{}
hours map[int]struct{}
daysOfMonth map[int]struct{}
months map[int]struct{}
daysOfWeek map[int]struct{}
anyDayOfMonth bool
anyDayOfWeek bool
location *time.Location
}
func parseCronSchedule(expression, timezone string) (cronSchedule, error) {
parts := strings.Fields(strings.TrimSpace(expression))
if len(parts) != 5 {
return cronSchedule{}, errors.New("cronExpression must contain five fields: minute hour day month weekday")
}
location, err := time.LoadLocation(strings.TrimSpace(timezone))
if err != nil {
return cronSchedule{}, fmt.Errorf("invalid schedule timezone %q: %w", timezone, err)
}
minutes, _, err := parseCronField(parts[0], 0, 59, false)
if err != nil {
return cronSchedule{}, fmt.Errorf("invalid cron minute: %w", err)
}
hours, _, err := parseCronField(parts[1], 0, 23, false)
if err != nil {
return cronSchedule{}, fmt.Errorf("invalid cron hour: %w", err)
}
daysOfMonth, anyDayOfMonth, err := parseCronField(parts[2], 1, 31, false)
if err != nil {
return cronSchedule{}, fmt.Errorf("invalid cron day: %w", err)
}
months, _, err := parseCronField(parts[3], 1, 12, false)
if err != nil {
return cronSchedule{}, fmt.Errorf("invalid cron month: %w", err)
}
daysOfWeek, anyDayOfWeek, err := parseCronField(parts[4], 0, 7, true)
if err != nil {
return cronSchedule{}, fmt.Errorf("invalid cron weekday: %w", err)
}
return cronSchedule{
minutes: minutes,
hours: hours,
daysOfMonth: daysOfMonth,
months: months,
daysOfWeek: daysOfWeek,
anyDayOfMonth: anyDayOfMonth,
anyDayOfWeek: anyDayOfWeek,
location: location,
}, nil
}
func nextCronTime(expression, timezone string, after time.Time) (time.Time, error) {
schedule, err := parseCronSchedule(expression, timezone)
if err != nil {
return time.Time{}, err
}
candidate := after.In(schedule.location).Truncate(time.Minute).Add(time.Minute)
const maxMinutes = 366 * 24 * 60 * 5
for checked := 0; checked < maxMinutes; checked++ {
if schedule.matches(candidate) {
return candidate, nil
}
candidate = candidate.Add(time.Minute)
}
return time.Time{}, errors.New("cronExpression has no execution time within five years")
}
func (schedule cronSchedule) matches(candidate time.Time) bool {
if _, ok := schedule.minutes[candidate.Minute()]; !ok {
return false
}
if _, ok := schedule.hours[candidate.Hour()]; !ok {
return false
}
if _, ok := schedule.months[int(candidate.Month())]; !ok {
return false
}
_, dayMatches := schedule.daysOfMonth[candidate.Day()]
_, weekdayMatches := schedule.daysOfWeek[int(candidate.Weekday())]
switch {
case schedule.anyDayOfMonth && schedule.anyDayOfWeek:
return true
case schedule.anyDayOfMonth:
return weekdayMatches
case schedule.anyDayOfWeek:
return dayMatches
default:
return dayMatches || weekdayMatches
}
}
func parseCronField(spec string, minValue, maxValue int, normalizeSunday bool) (map[int]struct{}, bool, error) {
spec = strings.TrimSpace(spec)
if spec == "" {
return nil, false, errors.New("field is empty")
}
values := make(map[int]struct{})
any := spec == "*"
for _, item := range strings.Split(spec, ",") {
item = strings.TrimSpace(item)
if item == "" {
return nil, false, errors.New("field contains an empty list item")
}
step := 1
base := item
if slash := strings.IndexByte(item, '/'); slash >= 0 {
base = item[:slash]
parsedStep, err := strconv.Atoi(item[slash+1:])
if err != nil || parsedStep <= 0 {
return nil, false, fmt.Errorf("invalid step %q", item[slash+1:])
}
step = parsedStep
}
start, end := minValue, maxValue
switch {
case base == "*":
case strings.Contains(base, "-"):
bounds := strings.Split(base, "-")
if len(bounds) != 2 {
return nil, false, fmt.Errorf("invalid range %q", base)
}
var err error
start, err = strconv.Atoi(bounds[0])
if err != nil {
return nil, false, fmt.Errorf("invalid range start %q", bounds[0])
}
end, err = strconv.Atoi(bounds[1])
if err != nil {
return nil, false, fmt.Errorf("invalid range end %q", bounds[1])
}
default:
value, err := strconv.Atoi(base)
if err != nil {
return nil, false, fmt.Errorf("invalid value %q", base)
}
start, end = value, value
}
if start < minValue || end > maxValue || start > end {
return nil, false, fmt.Errorf("value %d-%d is outside %d-%d", start, end, minValue, maxValue)
}
for value := start; value <= end; value += step {
if normalizeSunday && value == 7 {
values[0] = struct{}{}
} else {
values[value] = struct{}{}
}
}
}
return values, any, nil
}

View File

@@ -0,0 +1,105 @@
package syncjob
import (
"strings"
"testing"
"time"
)
func validValidationTestDefinition() JobDefinition {
return JobDefinition{
Name: "orders sync",
Lifecycle: JobLifecycleReady,
Kind: JobKindReconcile,
IncrementalMode: IncrementalSnapshot,
Source: EndpointRef{ConnectionID: "source"},
Target: EndpointRef{ConnectionID: "target"},
Mappings: []TableMapping{{
SourceTable: "orders",
TargetTable: "orders_archive",
Enabled: true,
}},
}
}
func TestValidateDefinitionRejectsUnsupportedPerMappingTargetStrategy(t *testing.T) {
definition := validValidationTestDefinition()
definition.Mappings[0].TargetTableStrategy = "drop_and_replace"
err := ValidateDefinition(definition)
if err == nil || !strings.Contains(err.Error(), "targetTableStrategy") {
t.Fatalf("ValidateDefinition error = %v, want targetTableStrategy error", err)
}
}
func TestNormalizeDefinitionPreservesDisabledMapping(t *testing.T) {
definition := validValidationTestDefinition()
definition.Mappings[0].Enabled = false
normalized := NormalizeDefinition(definition)
if normalized.Mappings[0].Enabled {
t.Fatal("normalization must not re-enable an explicitly disabled mapping")
}
}
func TestValidateDefinitionIgnoresIncompleteDisabledMapping(t *testing.T) {
definition := validValidationTestDefinition()
definition.Mappings = append(definition.Mappings, TableMapping{Enabled: false})
if err := ValidateDefinition(definition); err != nil {
t.Fatalf("disabled draft mapping was rejected: %v", err)
}
}
func TestValidateDefinitionAllowsQuerySinkWithoutSyntheticSourceTable(t *testing.T) {
definition := validValidationTestDefinition()
definition.Kind = JobKindQuerySink
definition.SourceQuery = "SELECT id, total FROM orders WHERE exported = false"
definition.Mappings[0].SourceTable = ""
if err := ValidateDefinition(definition); err != nil {
t.Fatalf("query sink definition was rejected: %v", err)
}
}
func TestValidateDefinitionRejectsIncrementalCompare(t *testing.T) {
definition := validValidationTestDefinition()
definition.Kind = JobKindCompare
definition.IncrementalMode = IncrementalWatermark
definition.Mappings[0].Watermark = &WatermarkSpec{Column: "updated_at", TieBreakerColumns: []string{"id"}}
err := ValidateDefinition(definition)
if err == nil || !strings.Contains(err.Error(), "only support snapshot") {
t.Fatalf("ValidateDefinition error = %v, want compare snapshot-only error", err)
}
}
func TestContinuousCDCJobsAreScheduledAndForbidOverlap(t *testing.T) {
definition := validValidationTestDefinition()
definition.Lifecycle = JobLifecycleEnabled
definition.IncrementalMode = IncrementalCDC
definition.CDC = &CDCSpec{Adapter: "mongodb-change-stream", StartPosition: "checkpoint"}
definition.Mappings[0].KeyColumns = []string{"id"}
definition.Schedule = ScheduleSpec{Kind: ScheduleContinuous}
definition.ConcurrencyPolicy = "forbid"
if err := ValidateDefinition(definition); err != nil {
t.Fatalf("continuous CDC definition was rejected: %v", err)
}
after := time.UnixMilli(1_700_000_000_000)
if got, want := NextRunAt(definition, after), after.Add(continuousRunPoll).UnixMilli(); got != want {
t.Fatalf("NextRunAt() = %d, want %d", got, want)
}
definition.ConcurrencyPolicy = "queue"
err := ValidateDefinition(definition)
if err == nil || !strings.Contains(err.Error(), "forbid concurrency") {
t.Fatalf("ValidateDefinition error = %v, want continuous overlap error", err)
}
}
func TestValidateDefinitionRejectsCDCWithoutStableKeys(t *testing.T) {
definition := validValidationTestDefinition()
definition.IncrementalMode = IncrementalCDC
definition.CDC = &CDCSpec{Adapter: "mongodb-change-stream", StartPosition: "latest"}
err := ValidateDefinition(definition)
if err == nil || !strings.Contains(err.Error(), "stable keyColumns") {
t.Fatalf("ValidateDefinition error = %v, want CDC key error", err)
}
}