Skip to content

Fix: Sync write path, CDC backpressure, durable offsets, security, workflow/DLQ/retry - #54

Merged
lekhrocks merged 13 commits into
mainfrom
fix/sync-cdc-security-workflow
Aug 9, 2026
Merged

Fix: Sync write path, CDC backpressure, durable offsets, security, workflow/DLQ/retry#54
lekhrocks merged 13 commits into
mainfrom
fix/sync-cdc-security-workflow

Conversation

@lekhrocks

@lekhrocks lekhrocks commented Aug 9, 2026

Copy link
Copy Markdown
Owner

Summary

This PR fixes critical bugs and architectural issues across the SyncFlow CDC platform:

🔧 Core Fixes

  1. Sync Write Path (3e135ad)

    • JdbcBatchWriter: Fixed currentTable/currentColumns NPE (was INSERT INTO null)
    • DestinationRouter: Batched writes per pipeline, periodic flush/commit instead of per-event connect/commit/close
    • SyncOrchestrator: Routes by pipelineId, closes destination writer on stop

  2. CDC Backpressure (3e135ad)

    • BoundedQueueEventPublisher: Blocking put() instead of drop-oldest — backpressure instead of silent data loss

  3. Durable Debezium Offsets (a20730c, 5955834)

    • JdbcOffsetBackingStore: Postgres-backed OffsetBackingStore (V13 migration) replaces /tmp file store
    • Offsets survive pod restarts — no re-processing or missed events
    • Pipeline-scoped slot/publication names; slot.drop.on.stop=true

  4. Security (1cc5398)

    • Removed committed AES/JWT defaults; fail-fast validation
    • Principal-based tenant scoping (JWT tid/oid/wid/pid claims)
    • All 10 controllers gate via AuthorizationService
    • Agent endpoints: fail-closed shared-secret guard

  5. Workflow/DLQ/Retry (6b32201)

    • WorkflowScheduler: Executes ready tasks, records TaskExecution, advances DAG
    • DeadLetterQueue.replay(): Re-enqueues stored event to sync engine
    • RetryEngine: Actual re-delivery after exponential backoff with tenant-aware callback

  6. Runtime State Persistence (cc44a3a, 2531e9d)

    • V12 migration: 8 tables (snapshot_jobs, sync_jobs, workflow_instances, quotas, audit_records, api_keys, agents, alert_events + checkpoints)
    • All services rewired to JPA; transient maps stay in-memory
    • SnapshotExecutor: fixes job leak (remove on complete/fail/cancel)

  7. Orphan Removal (6faf19e): Removed shadow PipelineService/PipelineEntity family + 3 DTOs

  8. SSE (8d4b721): 5-min timeout + 30s dead-subscriber sweep

🧪 Testing

  • 488 tests pass (0 failures, 11 skipped = Docker-gated)
  • Spotless clean
  • Full compile green

Migration Notes

  • V12: Runtime state tables (tenant-scoped)
  • V13: debezium_offsets table for durable Debezium offsets
  • No breaking API changes; all configs via env vars (SYNCFLOW_ENCRYPTION_KEY, SYNCFLOW_JWT_SECRET, SYNCFLOW_AGENT_TOKEN)

@lekhrocks
lekhrocks force-pushed the fix/sync-cdc-security-workflow branch 2 times, most recently from d83d6d7 to 7e74e45 Compare August 9, 2026 09:13
lekhrocks added 13 commits August 9, 2026 15:13
@lekhrocks
refactor(core): remove orphaned pipeline model family
cda2852 
Delete the shadowed core Pipeline path (PipelineService,
InMemoryPipelineRepository, PipelineRepository, model Pipeline/
PipelineEvent/PipelineStatus) and the unused api PipelineEntity family
(PipelineEntity, PipelineJpaRepository, PipelineRepositoryAdapter,
PipelineEntityMapper) plus orphaned DTOs (CreatePipelineRequest,
PipelineResponse, UpdatePipelineRequest). The live pipeline path is
PipelineDesignerService + PipelineDesignEntity. Leave the `pipelines`
DB table intact for Flyway checksum compatibility.
@lekhrocks
fix(connector): repair sync write path and add CDC backpressure
fee1313 
- JdbcBatchWriter: set currentTable/currentColumns from writeBatch args
  (was null -> "INSERT INTO null"), flush on table/column switch, reset
  state on connect.
- DestinationRouter: keep one writer+connection per pipeline, buffer
  events, flush+commit on batch-size/time intervals instead of
  connect/commit/close per event; closePipeline on stop/run-end.
- SyncOrchestrator: route by pipelineId, close the destination writer on
  stop and normal run end.
- BoundedQueueEventPublisher: block (put) on full queue instead of
  dropping the oldest event — backpressure, never silent data loss.
@lekhrocks
fix(connector): durable Debezium offsets and pipeline-scoped replicat…
c4acb90 
…ion slots

- JdbcOffsetBackingStore: Postgres-backed OffsetBackingStore persisting
  Debezium offsets in a debezium_offsets table (Flyway V13) instead of the
  ephemeral /tmp file store — offsets survive pod restarts, preventing
  re-processing or missed events. Plain JDBC, no JPA, so the connector
  module stays Spring-Data-free.
- DebeziumCdcConnector: use the JDBC offset store, add jdbcUrl() hook.
- PostgresCdcConnector/MySqlCdcConnector: override jdbcUrl().
- PostgresCdcConnector: scope slot/publication names per pipeline (via
  runtimeProperties pipelineId) so two pipelines on one DB don't collide;
  slot.drop.on.stop=true so slots don't leak on the source.
@lekhrocks
fix(workflow, sync): make DAG progress, DLQ replay re-enqueue, retrie…
fed0a34 
…s re-deliver

- WorkflowInstance: add withExecution/completed/completedTaskIds so task
  outcomes drive DAG progression.
- WorkflowScheduler: execute ready tasks, record TaskExecution (RUNNING/
  COMPLETED/FAILED), advance the DAG, mark workflow COMPLETED. Previously
  completedTaskIds returned Set.of() so the graph never progressed.
- DeadLetterQueue.replay: re-enqueue the stored CDCEvent into the sync
  engine (was a flag-only no-op). @lazy SyncOrchestrator breaks the
  constructor cycle.
- RetryEngine: actually re-deliver transient failures after exponential
  backoff via a tenant-aware re-enqueue callback (was count-only), wired
  from SyncOrchestrator.
@lekhrocks
fix(sse): 5-minute server timeout and periodic dead-subscriber sweep
feade32 
StatusBroadcaster had no timeout (0) and only cleaned emitters on emit
error, so dead SSE connections accumulated. Add a 5-minute server timeout
plus a 30s scheduled sweep that drops emitters whose connection is dead
(probed via keepalive ping).
@lekhrocks
feat(core): persist runtime state to Postgres (snapshot/sync/workflow…
eef8a56 
…/ops)

Migrate in-memory ConcurrentHashMap runtime state to JPA + Flyway (V12):
snapshot_jobs, snapshot_checkpoints, sync_jobs, workflow_instances,
quotas, audit_records, api_keys, agents, alert_events — all tenant-scoped.
SnapshotExecutor, CheckpointStore, SyncOrchestrator, WorkflowScheduler,
QuotaEngine, EnterpriseAuditStore, ApiKeyStore, FleetManager, AlertEngine
round-trip durable state through JPA repositories; transient maps (event
queues, worker threads, running flags, leader state) stay in-memory.
SnapshotExecutor now removes jobs on complete/fail/cancel (fixes leak).
@lekhrocks
feat(core): add JPA entities/repos + Flyway V12 for runtime state
6cd0700 
New entity + repository classes backing the persisted runtime state
(snapshot/sync/workflow/quota/audit/apikey/agent/alert) and the V12
migration. Complement to the service rewiring.
@lekhrocks
security: fail-fast secrets, principal-based tenancy, controller auth…
78e9216 
…z, agent token

- EncryptionService validates the AES key (16/24/32 bytes) and fails fast;
  application.yml no longer ships committed default keys/secrets.
  docker-compose + k8s wire dev/prod secrets via env.
- TenantJwtAuthenticationConverter + TenantFilter: tenant derived from the
  authenticated principal (tid/oid/wid/pid claims), never trusted from
  X-Tenant-Id headers. Header honored only when it matches the principal.
- AuthService: JWT now carries the tid claim.
- All controllers (Connection/Pipeline/Sync/Snapshot/Capture/Metadata/
  Plugin/Workflow/Agent/EnterpriseAi) gate mutations/reads via
  AuthorizationService.require(ResourcePermission.*).
- AgentTokenFilter: fail-closed shared-secret guard for agent register/
  heartbeat; agent management endpoints stay authenticated.
- ConnectionMapper/ConnectionService: inject the shared ObjectMapper
  instead of ad-hoc new ObjectMapper().
@lekhrocks
style: apply spotless formatting to integrated changes
8179ec0
@lekhrocks
fix(connector): durable Debezium offsets and pipeline-scoped replicat…
bbab5fb 
…ion slots

- JdbcOffsetBackingStore: Postgres-backed OffsetBackingStore persisting
  Debezium offsets in a debezium_offsets table (Flyway V13) instead of the
  ephemeral /tmp file store — offsets survive pod restarts, preventing
  re-processing or missed events. Plain JDBC, no JPA, so the connector
  module stays Spring-Data-free.
- DebeziumCdcConnector: use the JDBC offset store, add jdbcUrl() hook.
- PostgresCdcConnector/MySqlCdcConnector: override jdbcUrl().
- PostgresCdcConnector: scope slot/publication names per pipeline (via
  runtimeProperties pipelineId) so two pipelines on one DB don't collide;
  slot.drop.on.stop=true so slots don't leak on the source.
@lekhrocks
style: apply spotless formatting to offset store changes
021232c
@lekhrocks
fix(cdc): configure JDBC offset store for integration tests
17fd069 
- CdcIntegrationTest: configure offset.storage.jdbc.* properties for JdbcOffsetBackingStore
- PostgresCdcConnector/MySqlCdcConnector: override jdbcUrl() for durable offset store
- MySqlCdcConnector: add jdbcUrl() override for MySQL JDBC URL
@lekhrocks
fix(workflow): reset() clears workflows from database for test isolation
c62fe3b 
The reset() method now clears all workflows from the database via
repository.deleteAll() in addition to resetting the leader flag.
This ensures test isolation when multiple tests create workflows.
@lekhrocks
lekhrocks force-pushed the fix/sync-cdc-security-workflow branch from 5730449 to c62fe3b Compare August 9, 2026 09:43
@lekhrocks
lekhrocks merged commit ff8fc9e into main Aug 9, 2026
18 checks passed
@lekhrocks
lekhrocks deleted the fix/sync-cdc-security-workflow branch August 9, 2026 10:00
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

No reviews

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

1 participant

@lekhrocks