Replace queue fanout with party-0 coordinator

.test.env

@@ -12,7 +12,6 @@ SMPC__GRAPH_CHECKPOINT_BUCKET_NAME=wf-smpcv2-dev-hnsw-checkpoint
 SMPC__MATCH_DISTANCES_BUFFER_SIZE=32
 SMPC__OVERRIDE_MAX_BATCH_SIZE=false
 SMPC__MAX_DB_SIZE=10000
-SMPC__BATCH_POLLING_TIMEOUT_SECS=1
 SMPC__HAWK_REQUEST_PARALLELISM=1
 SMPC__DISABLE_PERSISTENCE=false
 SMPC__AWS__REGION=us-east-1

Dockerfile

@@ -25,6 +25,13 @@ RUN rustup component add cargo
 RUN cargo install cargo-build-deps \
     && cargo install cargo-edit --version 0.13.6 --locked
 
+RUN curl -o /tmp/ca_smpcv2_party_stage_0.crt https://wf-smpcv2-ca-stage-eu-north-1.s3.eu-north-1.amazonaws.com/ca_party_stage_1.stage.smpcv2.worldcoin.dev.pem && \
+    curl -o /tmp/ca_smpcv2_party_stage_1.crt https://wf-smpcv2-ca-stage-eu-north-1.s3.eu-north-1.amazonaws.com/ca_party_stage_2.stage.smpcv2.worldcoin.dev.pem && \
+    curl -o /tmp/ca_smpcv2_party_stage_2.crt https://wf-smpcv2-ca-stage-eu-north-1.s3.eu-north-1.amazonaws.com/ca_party_stage_3.stage.smpcv2.worldcoin.dev.pem && \
+    curl -o /tmp/ca_smpcv2_party_prod_0.crt https://wf-smpcv2-ca-prod-eu-north-1.s3.eu-north-1.amazonaws.com/ca_party_prod_1.smpcv2.worldcoin.org.pem && \
+    curl -o /tmp/ca_smpcv2_party_prod_1.crt https://wf-smpcv2-ca-prod-eu-north-1.s3.eu-north-1.amazonaws.com/ca_party_prod_2.smpcv2.worldcoin.org.pem && \
+    curl -o /tmp/ca_smpcv2_party_prod_2.crt https://wf-smpcv2-ca-prod-eu-north-1.s3.eu-north-1.amazonaws.com/ca_party_prod_3.smpcv2.worldcoin.org.pem
+
 FROM --platform=linux/amd64 build-image as build-app
 WORKDIR /src/gpu-iris-mpc
 COPY . .
@@ -40,6 +47,7 @@ COPY --from=build-app /src/gpu-iris-mpc/target/x86_64-unknown-linux-gnu/release/
 COPY --from=build-app /src/gpu-iris-mpc/target/x86_64-unknown-linux-gnu/release/key-manager /bin/key-manager
 COPY --from=build-app /src/gpu-iris-mpc/target/x86_64-unknown-linux-gnu/release/reshare-server /bin/reshare-server
 COPY --from=build-app /src/gpu-iris-mpc/target/x86_64-unknown-linux-gnu/release/reshare-client /bin/reshare-client
+COPY --from=build-image /tmp/ca_smpcv2_party_*.crt /usr/local/share/ca-certificates/
 
 USER 65534
 ENTRYPOINT ["/bin/iris-mpc-gpu"]

deploy/dev/common-values-ampc-hnsw.yaml

@@ -33,7 +33,6 @@ env:
   SMPC__HNSW_SCHEMA_NAME_SUFFIX: "_minfhd"
   SMPC__MAX_DB_SIZE: "2000000"
   SMPC__MAX_BATCH_SIZE: "96"
-  SMPC__BATCH_SYNC_POLLING_TIMEOUT_SECS: "30"
   SMPC__PROCESSING_TIMEOUT_SECS: "1800"
   SMPC__HNSW_MIN_LAYER_SEARCH_BATCH_SIZE: "8000"
   SMPC__HAWK_REQUEST_PARALLELISM: "96"
@@ -48,6 +47,7 @@ env:
   SMPC__LUC_SERIAL_IDS_FROM_SMPC_REQUEST: "false"
   SMPC__AWS__REGION: "eu-central-1"
   SMPC__SERVICE_PORTS: '["4000","4001","4002"]'
+  SMPC__COORDINATOR_PORTS: '["4200","4201","4202"]'
   SMPC__SERVER_COORDINATION__IMAGE_NAME: $(IMAGE_NAME)
   SMPC__SHARES_BUCKET_NAME: "wf-smpcv2-dev-sns-requests-v2"
   SMPC__ENABLE_S3_IMPORTER: "false"
@@ -115,6 +115,15 @@ ports:
   - containerPort: 4102
     name: tcp-4102
     protocol: TCP
+  - containerPort: 4200
+    name: tcp-4200
+    protocol: TCP
+  - containerPort: 4201
+    name: tcp-4201
+    protocol: TCP
+  - containerPort: 4202
+    name: tcp-4202
+    protocol: TCP
 
 livenessProbe:
   httpGet:

deploy/prod/common-values-ampc-hnsw.yaml

@@ -17,7 +17,6 @@ env:
         name: application
   SMPC__MAX_DB_SIZE: "20000000"
   SMPC__MAX_BATCH_SIZE: "8"
-  SMPC__BATCH_SYNC_POLLING_TIMEOUT_SECS: "30"
   SMPC__PROCESSING_TIMEOUT_SECS: "1240"
   SMPC__HAWK_STREAM_PARALLELISM: "32"
   SMPC__DISABLE_PERSISTENCE: "false"
@@ -29,6 +28,7 @@ env:
   SMPC__LUC_SERIAL_IDS_FROM_SMPC_REQUEST: "false"
   SMPC__AWS__REGION: "eu-north-1"
   SMPC__SERVICE_PORTS: '["4000","4001","4002"]'
+  SMPC__COORDINATOR_PORTS: '["4200","4201","4202"]'
   SMPC__SERVER_COORDINATION__IMAGE_NAME: $(IMAGE_NAME)
   SMPC__SERVER_COORDINATION__STARTUP_SYNC_TIMEOUT_SECS: "3600"
   SMPC__SERVER_COORDINATION__HEARTBEAT_INITIAL_RETRIES: "65"
@@ -114,6 +114,15 @@ ports:
   - containerPort: 4102
     name: tcp-4102
     protocol: TCP
+  - containerPort: 4200
+    name: tcp-4200
+    protocol: TCP
+  - containerPort: 4201
+    name: tcp-4201
+    protocol: TCP
+  - containerPort: 4202
+    name: tcp-4202
+    protocol: TCP
 
 livenessProbe:
   httpGet:

deploy/prod/common-values-iris-mpc.yaml

@@ -34,6 +34,10 @@ env:
   SMPC__ANON_STATS_DATABASE__CREATE: "true"
   SMPC__ANON_STATS_DATABASE__LOAD_PARALLELISM: "8"
   SMPC__AWS__REGION: "eu-north-1"
+  SMPC__COORDINATOR_PORTS: '["4200","4201","4202"]'
+  SMPC__TLS__PRIVATE_KEY: "/etc/ssl/private/key.pem"
+  SMPC__TLS__LEAF_CERT: "/etc/ssl/private/certificate.crt"
+  SMPC__TLS__ROOT_CERTS: '["/usr/local/share/ca-certificates/ca_smpcv2_party_prod_0.crt", "/usr/local/share/ca-certificates/ca_smpcv2_party_prod_1.crt", "/usr/local/share/ca-certificates/ca_smpcv2_party_prod_2.crt"]'
   SMPC__REQUESTS_QUEUE_URL:
     valueFrom:
       secretKeyRef:
@@ -158,5 +162,9 @@ preStop:
   # preStop.sleepPeriod specifies the time spent in Terminating state before SIGTERM is sent
   sleepPeriod: 10
 
+mountSSLCerts:
+  enabled: true
+  mountPath: /etc/ssl/private
+
 # terminationGracePeriodSeconds specifies the grace time between SIGTERM and SIGKILL
 terminationGracePeriodSeconds: 240 # TODO: make it 3x SMPC__PROCESSING_TIMEOUT_SECS

deploy/stage/common-values-ampc-hnsw.yaml

@@ -27,10 +27,8 @@ env:
       secretKeyRef:
         key: DATABASE_AURORA_HNSW_URL
         name: application
-  SMPC__DB_BACKED_INGEST: "true" # POP-4051 DB-backed ingest (stage rollout 2026-07)
   SMPC__MAX_DB_SIZE: "2000000"
   SMPC__MAX_BATCH_SIZE: "1"
-  SMPC__BATCH_SYNC_POLLING_TIMEOUT_SECS: "30"
   SMPC__PROCESSING_TIMEOUT_SECS: "1240"  # 2 minutes per batch in stage, bump to 4 in prod
   SMPC__HAWK_REQUEST_PARALLELISM: "1024"
   SMPC__HAWK_CONNECTION_PARALLELISM: "48"
@@ -46,6 +44,7 @@ env:
   SMPC__LUC_SERIAL_IDS_FROM_SMPC_REQUEST: "false"
   SMPC__AWS__REGION: "eu-north-1"
   SMPC__SERVICE_PORTS: '["4000","4001","4002"]'
+  SMPC__COORDINATOR_PORTS: '["4200","4201","4202"]'
   SMPC__SERVER_COORDINATION__IMAGE_NAME: $(IMAGE_NAME)
   SMPC__SERVER_COORDINATION__HEARTBEAT_INITIAL_RETRIES: "65"
   SMPC__SERVER_COORDINATION__HEARTBEAT_INTERVAL_SECS: "3"
@@ -103,6 +102,15 @@ ports:
   - containerPort: 4002
     name: tcp-4002
     protocol: TCP
+  - containerPort: 4200
+    name: tcp-4200
+    protocol: TCP
+  - containerPort: 4201
+    name: tcp-4201
+    protocol: TCP
+  - containerPort: 4202
+    name: tcp-4202
+    protocol: TCP
 
 livenessProbe:
   httpGet:

deploy/stage/common-values-iris-mpc.yaml

@@ -21,7 +21,6 @@ env:
   SMPC__DATABASE__CREATE: "true"
   SMPC__DATABASE__LOAD_PARALLELISM: "8"
   SMPC__DISABLE_PERSISTENCE: "false"
-  SMPC__DB_BACKED_INGEST: "true" # POP-4051 DB-backed ingest (stage rollout 2026-07)
   SMPC__ANON_STATS_DATABASE__URL:
     valueFrom:
       secretKeyRef:
@@ -31,6 +30,10 @@ env:
   SMPC__ANON_STATS_DATABASE__CREATE: "false"
   SMPC__ANON_STATS_DATABASE__LOAD_PARALLELISM: "8"
   SMPC__AWS__REGION: "eu-north-1"
+  SMPC__COORDINATOR_PORTS: '["4200","4201","4202"]'
+  SMPC__TLS__PRIVATE_KEY: "/etc/ssl/private/key.pem"
+  SMPC__TLS__LEAF_CERT: "/etc/ssl/private/certificate.crt"
+  SMPC__TLS__ROOT_CERTS: '["/usr/local/share/ca-certificates/ca_smpcv2_party_stage_0.crt", "/usr/local/share/ca-certificates/ca_smpcv2_party_stage_1.crt", "/usr/local/share/ca-certificates/ca_smpcv2_party_stage_2.crt"]'
   SMPC__REQUESTS_QUEUE_URL:
     valueFrom:
       secretKeyRef:
@@ -151,5 +154,9 @@ preStop:
   # preStop.sleepPeriod specifies the time spent in Terminating state before SIGTERM is sent
   sleepPeriod: 10
 
+mountSSLCerts:
+  enabled: true
+  mountPath: /etc/ssl/private
+
 # terminationGracePeriodSeconds specifies the grace time between SIGTERM and SIGKILL
 terminationGracePeriodSeconds: 120

docs/architecture-overview.md

@@ -58,7 +58,8 @@ The server process that bridges AWS infrastructure with the MPC engine.
 
 | Module | Path | Role |
 |--------|------|------|
-| Batch processor | `src/services/processors/batch.rs` | Collects SQS messages into `BatchQuery` |
+| Coordinator | `src/coordinator.rs` | Party-0 FIFO ingress, API, and prepare/commit protocol |
+| Batch processor | `src/services/processors/batch.rs` | Converts coordinated envelopes into `BatchQuery` |
 | Job processor | `src/services/processors/job.rs` | Converts `ServerJobResult` to SNS responses, persists |
 | Modifications sync | `src/services/processors/modifications_sync.rs` | Rollforward/rollback of modifications |
 
@@ -74,16 +75,22 @@ Binaries and operational scripts.
 ## Data Flow (Simplified)
 
 ```
-Service Client                    Server                         MPC Engine
-─────────────                    ──────                         ──────────
-S3 upload shares  ──────►
-SNS publish request ────►  SQS receive ──► BatchQuery ────►  HawkActor.handle_job()
-                           collect batch     build batch        ├─ apply_deletions
-                                                                ├─ do_search (Normal)
-                                                                ├─ do_search (Mirror)
-                                                                ├─ handle_mutations
-                                                                └─ HawkResult
-                          SNS publish  ◄── ServerJobResult ◄── HawkResult.job_result()
-SQS receive response ◄──
-correlate with request
+Client                       Party 0 coordinator                 MPC parties
+──────                       ───────────────────                 ───────────
+POST request ──────────────► durable FIFO in Postgres
+legacy SQS (optional) ─────►
+                              Prepare over dedicated mTLS ─────► build BatchQuery
+                              ◄──────────── Prepared/digest
+                              Commit ──────────────────────────► execute GPU or HNSW job
+GET result ◄─────────────── completed/rejected/failed result
+legacy SNS (optional) ◄──── party-0 publish only
 ```
+
+The coordinator API is mounted on party 0's existing coordination HTTP server:
+
+- `POST /coordinator/requests` accepts `message_type`, `payload`, and an optional idempotency `request_id`.
+- `GET /coordinator/requests/{request_id}` returns `pending`, `preparing`, `processing`, `completed`, `rejected`, or `failed`.
+
+The request inbox and result are stored in Postgres rather than a local WAL, because server disks may be ephemeral. Inter-party coordinator traffic uses a separate connection from the execution layer while reusing the existing TCP/mTLS networking stack.
+
+During prepare, each party reports requests whose shares could not be fetched, parsed, or decrypted. Party 0 atomically persists the union as `rejected`, and the commit instructs every party to discard those requests before GPU or HNSW execution.

docs/batch-query.md

@@ -15,6 +15,8 @@ pub struct BatchQuery {
 
     // Uniqueness + Reauth + ResetCheck fields
     pub request_ids: Vec<String>,              // Operation IDs (signup_id, reauth_id, reset_id)
+    pub coordinator_request_ids: Vec<String>,  // Ledger IDs for matching requests
+    pub ordered_request_ids: Vec<String>,      // Ledger IDs across every request type
     pub request_types: Vec<String>,            // Message type strings
     pub metadata: Vec<BatchMetadata>,
     pub skip_persistence: Vec<bool>,
@@ -65,17 +67,17 @@ pub enum RequestIndex {
 ```rust
 // Adds uniqueness/reauth/reset_check request
 pub fn push_matching_request(
-    sns_message_id, request_id, request_type, metadata, or_rule_serial_ids, skip_persistence
+    coordinator_request_id, request_id, request_type, metadata, or_rule_serial_ids, skip_persistence
 )
 
 // Adds deletion request
 pub fn push_deletion_request(
-    sns_message_id, deletion_0_index: u32, metadata   // deletion_0_index is 0-based
+    coordinator_request_id, deletion_0_index: u32, metadata   // deletion_0_index is 0-based
 )
 
 // Adds reset update request
-pub fn push_reset_update_request(
-    sns_message_id, reset_update_0_index: u32, request_id, metadata, shares
+pub fn push_identity_update_request(
+    coordinator_request_id, request_id, request_type, identity_update_0_index: u32, shares
 )
 ```
 

docs/request-lifecycle.md

@@ -12,7 +12,10 @@ tbd
 
 **File:** `iris-mpc/src/services/processors/batch.rs`
 
-The server collects SQS messages into a `BatchQuery`:
+Party 0 accepts requests through the coordinator API or the compatibility SQS
+queue, assigns their FIFO order, and sends the ordered batch to the other
+parties over the dedicated mTLS coordinator connection.
+Each party then converts the same envelopes into a `BatchQuery`:
 
 | Message Type | Handler | Key Logic |
 |-------------|---------|-----------|
@@ -22,15 +25,17 @@ The server collects SQS messages into a `BatchQuery`:
 | `reset_check` | `process_reset_check_request` | Query-only, no persistence |
 | `reset_update` | `process_reset_update_request` | Iris replacement at target serial_id |
 
-**Deletion dedup** (lines 436-447): If the same serial_id already has a modification in the CURRENT batch, the second request is silently skipped (`return Ok(())`). The SQS message is already deleted (line 427), so no response is ever generated for the skipped request.
+If two modifications in one batch target the same serial ID, the later request
+is rejected before execution. Party 0 records the rejected coordinator row and
+publishes the compatibility error result; all parties discard the same request.
 
 This dedup only applies WITHIN a single server batch. Across batches, the `modifications` map is fresh.
 
 ### Job Result Processing
 
 **File:** `iris-mpc/src/services/processors/job.rs`
 
-After MPC computation, `ServerJobResult` is converted to per-request SNS responses:
+After MPC computation, `ServerJobResult` is converted to per-request results:
 
 ```
 identity_deletion_results = deleted_ids.iter().map(|idx| {
@@ -39,7 +44,8 @@ identity_deletion_results = deleted_ids.iter().map(|idx| {
 })
 ```
 
-Results sent to SNS per type (lines 389-440):
+Party 0 saves each result in `coordinator_requests` for API polling and, when a
+results topic is configured, publishes the same compatibility SNS events:
 1. Uniqueness results
 2. Reauth results
 3. Identity deletion results
@@ -87,6 +93,6 @@ MPC apply_deletions:
 ServerJobResult.deleted_ids = [0-based indices]
   ↓ job.rs: serial_id = idx + 1
 IdentityDeletionResult { serial_id }   →  back to 1-based
-  ↓ SNS → SQS
-Client is_correlation: parent.SerialId == result.serial_id
+  ↓ coordinator API polling or compatibility SNS → SQS
+Client correlation: parent.SerialId == result.serial_id
 ```

iris-mpc-bins/Cargo.toml

@@ -15,7 +15,6 @@ helpers = [
   "dep:reqwest",
 ]
 phase_trace = ["iris-mpc-cpu/phase_trace"]
-explicit-sns-batching = ["iris-mpc/explicit-sns-batching", "iris-mpc-common/explicit-sns-batching", "iris-mpc-utils/explicit-sns-batching"]
 networking_metrics = ["iris-mpc-cpu/networking_metrics"]
 
 [dependencies]