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

docker/docker-compose.yml

@@ -40,8 +40,12 @@ services:
       SPRING_DATASOURCE_USERNAME: syncflow
       SPRING_DATASOURCE_PASSWORD: syncflow
       SPRING_JPA_HIBERNATE_DDL_AUTO: validate
-      # JWT auth (must-change-password is on for admin-provisioned accounts).
-      SYNCFLOW_JWT_SECRET: ${SYNCFLOW_JWT_SECRET:-c3luY2Zsb3ctaHMyNTYtand0LXNlY3JldC1rZXktMjAyNi1jaGFuZ2UtaW4tcHJvZA==}
+      # ─── DEV-ONLY secrets. The app has no defaults (fail-fast) — supply real
+      #     values in non-local environments. NEVER use these in production.
+      #     AES-256 key = base64("dev-encryption-key-0123456789abc") (32 bytes).
+      SYNCFLOW_ENCRYPTION_KEY: ZGV2LWVuY3J5cHRpb24ta2V5LTAxMjM0NTY3ODlhYmM=
+      #     HS256 secret = base64 of a 44-byte dev-only HMAC key.
+      SYNCFLOW_JWT_SECRET: ZGV2LWp3dC1zZWNyZXQta2V5LWZvci1sb2NhbC1kZXYtb25seS1rZXktMDE=
       # Kafka transport (off by default).
       SYNCFLOW_KAFKA_ENABLED: ${SYNCFLOW_KAFKA_ENABLED:-false}
       SYNCFLOW_KAFKA_BOOTSTRAP_SERVERS: kafka:9092

k8s/base/deployment.yaml

@@ -59,6 +59,11 @@ spec:
                 secretKeyRef:
                   name: syncflow-encryption
                   key: key
+            - name: SYNCFLOW_JWT_SECRET
+              valueFrom:
+                secretKeyRef:
+                  name: syncflow-jwt
+                  key: secret
           volumeMounts:
             - name: config
               mountPath: /app/config

syncflow-api/src/main/java/com/syncflow/api/agent/AgentController.java

@@ -1,6 +1,8 @@
 package com.syncflow.api.agent;
 
 import com.syncflow.agent.domain.Agent;
+import com.syncflow.api.security.rbac.AuthorizationService;
+import com.syncflow.api.security.rbac.ResourcePermission;
 import com.syncflow.agent.domain.AgentId;
 import com.syncflow.agent.domain.HardwareMetrics;
 import org.springframework.http.ResponseEntity;
@@ -19,9 +21,11 @@
 public class AgentController {
 
     private final FleetManager fleetManager;
+    private final AuthorizationService authz;
 
-    public AgentController(FleetManager fleetManager) {
+    public AgentController(FleetManager fleetManager, AuthorizationService authz) {
         this.fleetManager = fleetManager;
+        this.authz = authz;
     }
 
     @PostMapping("/register")
@@ -57,29 +61,34 @@ public ResponseEntity<Map<String, Object>> heartbeat(@RequestBody Map<String, Ob
 
     @GetMapping
     public ResponseEntity<List<Agent>> list() {
+        authz.require(ResourcePermission.CONNECTION_READ);
         return ResponseEntity.ok(fleetManager.list());
     }
 
     @GetMapping("/{id}")
     public ResponseEntity<Agent> get(@PathVariable String id) {
+        authz.require(ResourcePermission.CONNECTION_READ);
         return fleetManager.get(new AgentId(id))
                 .map(ResponseEntity::ok)
                 .orElse(ResponseEntity.notFound().build());
     }
 
     @PostMapping("/{id}/drain")
     public ResponseEntity<Map<String, Object>> drain(@PathVariable String id) {
+        authz.require(ResourcePermission.PIPELINE_EXECUTE);
         fleetManager.drain(new AgentId(id));
         return ResponseEntity.ok(Map.of("agentId", id, "status", "DRAINING"));
     }
 
     @PostMapping("/{id}/restart")
     public ResponseEntity<Map<String, Object>> restart(@PathVariable String id) {
+        authz.require(ResourcePermission.PIPELINE_EXECUTE);
         return ResponseEntity.ok(Map.of("agentId", id, "action", "restart_requested"));
     }
 
     @GetMapping("/{id}/metrics")
     public ResponseEntity<Map<String, Object>> metrics(@PathVariable String id) {
+        authz.require(ResourcePermission.EXECUTION_READ);
         return fleetManager.get(new AgentId(id))
                 .map(a -> ResponseEntity.<Map<String, Object>>ok(Map.of(
                         "agentId", id,

syncflow-api/src/main/java/com/syncflow/api/agent/FleetManager.java

@@ -1,11 +1,18 @@
 package com.syncflow.api.agent;
 
+import com.fasterxml.jackson.core.type.TypeReference;
 import com.syncflow.agent.domain.Agent;
 import com.syncflow.agent.domain.AgentId;
 import com.syncflow.agent.domain.AgentStatus;
 import com.syncflow.agent.domain.HardwareMetrics;
+import com.syncflow.api.agent.entity.AgentEntity;
+import com.syncflow.api.agent.repository.AgentRepository;
 import com.syncflow.api.ops.metrics.MetricsRegistry;
+import com.syncflow.api.runtimestate.RuntimeStateJson;
+import com.syncflow.tenant.TenantSupport;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
 
 import java.time.Duration;
 import java.time.Instant;
@@ -19,14 +26,28 @@
 @Component
 public class FleetManager {
 
-    private final Map<AgentId, Agent> agents = new ConcurrentHashMap<>();
+    private final AgentRepository repository;
+    private final RuntimeStateJson json;
     private final MetricsRegistry metrics;
+
+    // Fast-path cache; durable source of truth is the agents table.
+    private final Map<AgentId, Agent> agents = new ConcurrentHashMap<>();
     private final AtomicLong agentCounter = new AtomicLong(0);
     private final Map<String, LongAdder> onlineByRegion = new ConcurrentHashMap<>();
 
     private static final Duration HEARTBEAT_TIMEOUT = Duration.ofSeconds(60);
 
+    @Autowired
+    public FleetManager(AgentRepository repository, RuntimeStateJson json, MetricsRegistry metrics) {
+        this.repository = repository;
+        this.json = json;
+        this.metrics = metrics;
+    }
+
+    /** Unit-test seam: in-memory fleet without a repository. */
     public FleetManager(MetricsRegistry metrics) {
+        this.repository = null;
+        this.json = null;
         this.metrics = metrics;
     }
 
@@ -45,45 +66,61 @@ private void decOnline(String region) {
             adder.decrement();
     }
 
+    @Transactional
     public Agent register(String version, List<String> capabilities,
             Map<String, String> labels, String environment,
             String region, String hostname) {
         var agent = Agent.register(version, capabilities, labels, environment, region, hostname);
         agents.put(agent.id(), agent);
         agentCounter.incrementAndGet();
         incOnline(region);
+        persist(agent);
         return agent;
     }
 
+    @Transactional
     public Optional<Agent> heartbeat(AgentId id, HardwareMetrics hw) {
         return Optional.ofNullable(agents.computeIfPresent(id, (k, agent) -> {
             var updated = agent.withHeartbeat(hw);
             pruneOffline();
+            persist(updated);
             return updated;
         }));
     }
 
+    @Transactional
     public void markOffline(AgentId id) {
         agents.computeIfPresent(id, (k, a) -> {
             if (a.status() == AgentStatus.ONLINE)
                 decOnline(a.region());
-            return a.withStatus(AgentStatus.OFFLINE);
+            var updated = a.withStatus(AgentStatus.OFFLINE);
+            persist(updated);
+            return updated;
         });
     }
 
+    @Transactional
     public void drain(AgentId id) {
         agents.computeIfPresent(id, (k, a) -> {
             if (a.status() == AgentStatus.ONLINE)
                 decOnline(a.region());
-            return a.withStatus(AgentStatus.DRAINING);
+            var updated = a.withStatus(AgentStatus.DRAINING);
+            persist(updated);
+            return updated;
         });
     }
 
+    @Transactional(readOnly = true)
     public Optional<Agent> get(AgentId id) {
         return Optional.ofNullable(agents.get(id));
     }
 
+    @Transactional(readOnly = true)
     public List<Agent> list() {
+        if (repository != null)
+            return repository.findByTenantId(TenantSupport.tenantId()).stream()
+                    .map(this::toDomain)
+                    .toList();
         return List.copyOf(agents.values());
     }
 
@@ -104,7 +141,42 @@ private void pruneOffline() {
                         && a.lastHeartbeat().isBefore(threshold))
                 .forEach(a -> {
                     decOnline(a.region());
-                    agents.put(a.id(), a.withStatus(AgentStatus.UNREACHABLE));
+                    var updated = a.withStatus(AgentStatus.UNREACHABLE);
+                    agents.put(a.id(), updated);
+                    persist(updated);
                 });
     }
+
+    private void persist(Agent agent) {
+        if (repository == null)
+            return; // unit-test seam
+        var entity = repository.findById(agent.id().value()).orElseGet(AgentEntity::new);
+        entity.setId(agent.id().value());
+        entity.setTenantId(TenantSupport.tenantId());
+        entity.setVersion(agent.version());
+        entity.setStatus(agent.status().name());
+        entity.setCapabilities(json.toJson(agent.capabilities()));
+        entity.setLabels(json.toJson(agent.labels()));
+        entity.setEnvironment(agent.environment());
+        entity.setRegion(agent.region());
+        entity.setHostname(agent.hostname());
+        entity.setHardware(json.toJson(agent.hardware()));
+        entity.setRegisteredAt(agent.registeredAt());
+        entity.setLastHeartbeat(agent.lastHeartbeat());
+        entity.setCreatedAt(agent.registeredAt());
+        entity.setUpdatedAt(Instant.now());
+        repository.save(entity);
+    }
+
+    private Agent toDomain(AgentEntity e) {
+        return new Agent(new AgentId(e.getId()), e.getVersion(),
+                AgentStatus.valueOf(e.getStatus()),
+                json.fromJson(e.getCapabilities(), new TypeReference<List<String>>() {
+                }),
+                json.fromJson(e.getLabels(), new TypeReference<Map<String, String>>() {
+                }),
+                e.getEnvironment(), e.getRegion(), e.getHostname(),
+                json.fromJson(e.getHardware(), HardwareMetrics.class),
+                e.getRegisteredAt(), e.getLastHeartbeat());
+    }
 }

syncflow-api/src/main/java/com/syncflow/api/agent/entity/AgentEntity.java

@@ -0,0 +1,69 @@
+package com.syncflow.api.agent.entity;
+
+import jakarta.persistence.Column;
+import jakarta.persistence.Entity;
+import jakarta.persistence.Id;
+import jakarta.persistence.Table;
+import lombok.Getter;
+import lombok.Setter;
+import org.hibernate.annotations.JdbcTypeCode;
+import org.hibernate.type.SqlTypes;
+
+import java.time.Instant;
+
+/** Fleet agent; capabilities, labels, hardware metrics as JSONB. */
+@Setter
+@Getter
+@Entity
+@Table(name = "agents")
+public class AgentEntity {
+
+    @Id
+    @Column(length = 36)
+    private String id;
+
+    @Column(name = "tenant_id", nullable = false, length = 36)
+    private String tenantId = "00000000-0000-0000-0000-000000000000";
+
+    @Column(length = 50)
+    private String version;
+
+    @Column(nullable = false, length = 20)
+    private String status;
+
+    @Column(nullable = false, columnDefinition = "jsonb")
+    @JdbcTypeCode(SqlTypes.JSON)
+    private String capabilities;
+
+    @Column(nullable = false, columnDefinition = "jsonb")
+    @JdbcTypeCode(SqlTypes.JSON)
+    private String labels;
+
+    @Column(length = 50)
+    private String environment;
+
+    @Column(length = 50)
+    private String region;
+
+    @Column(length = 255)
+    private String hostname;
+
+    @Column(nullable = false, columnDefinition = "jsonb")
+    @JdbcTypeCode(SqlTypes.JSON)
+    private String hardware;
+
+    @Column(name = "registered_at", nullable = false)
+    private Instant registeredAt;
+
+    @Column(name = "last_heartbeat", nullable = false)
+    private Instant lastHeartbeat;
+
+    @Column(name = "created_at", nullable = false)
+    private Instant createdAt;
+
+    @Column(name = "updated_at", nullable = false)
+    private Instant updatedAt;
+
+    public AgentEntity() {
+    }
+}

syncflow-api/src/main/java/com/syncflow/api/agent/repository/AgentRepository.java

@@ -0,0 +1,11 @@
+package com.syncflow.api.agent.repository;
+
+import com.syncflow.api.agent.entity.AgentEntity;
+import org.springframework.data.jpa.repository.JpaRepository;
+
+import java.util.List;
+
+public interface AgentRepository extends JpaRepository<AgentEntity, String> {
+
+    List<AgentEntity> findByTenantId(String tenantId);
+}

syncflow-api/src/main/java/com/syncflow/api/config/AuthSecurityBeans.java

@@ -2,19 +2,23 @@
 
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.core.convert.converter.Converter;
 import org.springframework.security.authentication.AuthenticationManager;
 import org.springframework.security.authentication.ProviderManager;
 import org.springframework.security.authentication.dao.DaoAuthenticationProvider;
 import org.springframework.security.core.userdetails.UserDetailsService;
 import org.springframework.security.crypto.bcrypt.BCryptPasswordEncoder;
 import org.springframework.security.crypto.password.PasswordEncoder;
-import org.springframework.security.oauth2.server.resource.authentication.JwtAuthenticationConverter;
+import org.springframework.security.oauth2.jwt.Jwt;
+import org.springframework.security.authentication.AbstractAuthenticationToken;
 
 /**
  * Auth beans: BCrypt password encoder, the AuthenticationManager backed by
  * DaoAuthenticationProvider over the user-details service, and the JWT
  * authentication converter that maps the JWT {@code scope} claim to
- * {@code ROLE_*} authorities (consumed by TenantFilter / RBAC).
+ * {@code SCOPE_*} authorities and carries the caller's tenant claims (read by
+ * TenantFilter / RBAC — tenant is taken from the principal, not client
+ * headers).
  */
 @Configuration
 public class AuthSecurityBeans {
@@ -34,9 +38,9 @@ public AuthenticationManager authenticationManager(
     }
 
     @Bean
-    public JwtAuthenticationConverter jwtAuthenticationConverter() {
-        // Map JWT 'scope' claim -> ROLE_ authorities. The login token encodes the
-        // user's roles into 'scope'; TenantFilter reads authorities for RBAC.
-        return new JwtAuthenticationConverter();
+    public Converter<Jwt, AbstractAuthenticationToken> jwtAuthenticationConverter() {
+        // Maps JWT 'scope' -> SCOPE_ authorities and attaches the tenant scope
+        // (tid/oid/wid/pid claims) to the token details.
+        return new TenantJwtAuthenticationConverter();
     }
 }