Split batcher and ipfs publisher

.env.example

@@ -29,12 +29,16 @@ HEARTBEAT_TRACKER_SOURCE=heartbeat-tracker
 HEARTBEAT_TRACKER_HEALTH_PORT=3030
 HEARTBEAT_TRACKER_ONLINE_WINDOW_MS=30000
 HEARTBEAT_TRACKER_REDIS_PREFIX=heartbeat-tracker:v1
-# IPFS publisher (batches authorized telemetry to IPFS)
+# Batcher (batches authorized telemetry into telemetry.batched.v1)
+BATCHER_GROUP_ID=batcher-v1
+BATCHER_SOURCE=batcher
+BATCHER_HEALTH_PORT=3041
+BATCHER_BATCH_SIZE=10
+BATCHER_BATCH_TIMEOUT_MS=30000
+# IPFS publisher (publishes telemetry.batched.v1 batches to IPFS)
 IPFS_PUBLISHER_GROUP_ID=ipfs-publisher-v1
 IPFS_PUBLISHER_SOURCE=ipfs-publisher
 IPFS_PUBLISHER_HEALTH_PORT=3040
-IPFS_PUBLISHER_BATCH_SIZE=10
-IPFS_PUBLISHER_BATCH_TIMEOUT_MS=30000
 IPFS_PUBLISHER_ENABLE_COMPRESSION=true
 # Blockchain anchor (anchors IPFS CIDs to Robonomics CPS pallet)
 BLOCKCHAIN_ANCHOR_GROUP_ID=blockchain-anchor-v1

.github/copilot-instructions.md

@@ -39,7 +39,9 @@ This is an **event-driven telemetry pipeline** with Kafka as the central durable
 4. Multiple **downstream consumers** process authorized events independently:
    - **pubsub-broadcaster**: Publishes to libp2p/GossipSub for real-time web UI
    - **heartbeat-tracker**: Tracks sensor liveness/uptime metrics (observability-only, no DLQ)
-   - **ipfs-publisher**: Batches and publishes to IPFS, emits CIDs
+   - **batcher**: Batches authorized telemetry and emits `telemetry.batched.v1`
+5. **batcher** → **ipfs-publisher** → **blockchain-anchor** publication chain:
+   - **ipfs-publisher**: Consumes `telemetry.batched.v1`, publishes batches to IPFS, emits CIDs
    - **blockchain-anchor**: Anchors IPFS CIDs to Robonomics blockchain
 
 ### Key Architectural Constraints
@@ -58,8 +60,8 @@ Robonomics Blockchain → registry-sync → Redis → endpoint (lookup during va
 ### Core Kafka Topics
 - `telemetry.authorized.v1` - Successfully validated telemetry
 - `telemetry.rejected.v1` - Failed validation (signature/timestamp/auth)
-- `telemetry.pubsub.result.v1` - PubSub publish results
-- `telemetry.ipfs.result.v1` - IPFS publish results (includes CID)
+- `telemetry.batched.v1` - Batched authorized telemetry ready for publication
+- `ipfs.published.v1` - IPFS publish results (includes CID)
 - `telemetry.blockchain.result.v1` - Blockchain anchoring results
 - `telemetry.retry.v1` - Transient failures for retry
 - `telemetry.dlq.v1` - Exhausted retries (dead letters)
@@ -71,6 +73,7 @@ Robonomics Blockchain → registry-sync → Redis → endpoint (lookup during va
 - `services/whitelist` - Whitelist-based sensor auth provider
 - `services/pubsub-broadcaster` - Kafka→libp2p GossipSub bridge
 - `services/heartbeat-tracker` - Observability metrics (online sensors, uptime)
+- `services/batcher` - Kafka batcher (authorized → `telemetry.batched.v1`)
 - `services/ipfs-publisher` - Kafka→IPFS publisher (Kubo RPC)
 - `services/blockchain-anchor` - IPFS CID→blockchain anchoring
 - `tools/fake-sensor-cli` - Generate test telemetry with Ed25519 signatures

README.md

@@ -35,7 +35,8 @@ services/registry-sync     # Robonomics blockchain→Redis projection sync servi
 services/whitelist         # Whitelist-based sensor authentication provider
 services/pubsub-broadcaster # Kafka→libp2p GossipSub bridge for real-time web UI
 services/heartbeat-tracker  # Observability: sensor liveness & uptime metrics
-services/ipfs-publisher    # Kafka→IPFS publisher (batches, produces CIDs)
+services/batcher           # Kafka: batches authorized telemetry into telemetry.batched.v1
+services/ipfs-publisher    # Kafka→IPFS publisher (publishes batches, produces CIDs)
 services/blockchain-anchor # IPFS CID→Robonomics CPS pallet anchoring
 tools/fake-sensor-cli      # Generate test telemetry with Ed25519 signatures
 ```
@@ -93,5 +94,6 @@ Use `REJECTION_CODES` from `@scp/core` to reference these codes in your code:
 - **whitelist**: Alternative authentication provider - maintains static sensor whitelist in Redis, bypassing blockchain dependency for simpler deployments.
 - **pubsub-broadcaster**: Consumes `telemetry.authorized.v1`, publishes to libp2p/GossipSub for real-time web UI, emits `telemetry.pubsub.result.v1`, routes exhausted failures to DLQ.
 - **heartbeat-tracker**: Observability-only consumer of `telemetry.authorized.v1`, tracks sensor liveness (`firstSeen`, `lastSeen`, `onlineSince`) in Redis, exposes `sensors_online` count and per-sensor/aggregate uptime metrics over configurable online window (default 30s). Does not emit result events or participate in retry/DLQ.
-- **ipfs-publisher**: Consumes `telemetry.authorized.v1`, batches and publishes to IPFS, emits `telemetry.ipfs.result.v1`.
-- **blockchain-anchor**: Consumes `telemetry.ipfs.result.v1`, deduplicates by CID, emits `telemetry.blockchain.result.v1`.
+- **batcher**: Consumes `telemetry.authorized.v1`, groups events into deterministic batches (by size, lag, and a bounded flush timer), and emits `telemetry.batched.v1`. Serializes flushes (single active flush per instance) and flushes pending batches on graceful shutdown.
+- **ipfs-publisher**: Consumes `telemetry.batched.v1`, publishes/pins each batch to IPFS (optionally XZ-compressed), deduplicates by `batch_id`, emits `ipfs.published.v1`.
+- **blockchain-anchor**: Consumes `ipfs.published.v1`, deduplicates by CID, emits `telemetry.blockchain.result.v1`.

docker/kafka-init-topics.sh

@@ -5,6 +5,7 @@ set -e
 TOPICS=(
   "telemetry.authorized.v1"
   "telemetry.rejected.v1"
+  "telemetry.batched.v1"
   "ipfs.published.v1"
   "telemetry.dlq.v1"
 )

docs/architecture/project-architecture.md

@@ -20,7 +20,7 @@ The system accepts Ed25519-signed environmental sensor telemetry (Altruist-serie
 
 ### Blockchain anchoring
 
-`Trusted Messages (Kafka) -> IPFS Publisher -> Message Bus (Kafka) -> Robonomics Blockchain`
+`Trusted Messages (Kafka) -> Batcher -> Message Bus (Kafka) -> IPFS Publisher -> Message Bus (Kafka) -> Robonomics Blockchain`
 
 ## Module responsibilities
 
@@ -65,11 +65,20 @@ The system accepts Ed25519-signed environmental sensor telemetry (Altruist-serie
 - Does not emit result events and does not participate in retry/DLQ commit-result semantics.
 - Fault isolation: failures in heartbeat tracking do not block telemetry pipeline.
 
+### Batcher
+- Consumes authorized events from `telemetry.authorized.v1`.
+- Groups events into deterministic batches by size, consumer lag, and a bounded flush timer.
+- Serializes each batch as a `crypto.v1.SignedEnvelopeBatch` and emits `telemetry.batched.v1`.
+- Serializes flushes (single active flush per instance) so a timer-triggered flush cannot publish the same batch as a size/lag-triggered flush.
+- Flushes any pending batch during graceful shutdown before closing resources.
+- Commits `telemetry.authorized.v1` offsets only after the batch is durably produced.
+
 ### IPFS Publisher
-- Consumes authorized events from Kafka.
-- Batches, produces IPFS object/CAR, publishes CID.
-- Emits `telemetry.ipfs.result.v1`.
-- Commits offset only after publish/pin success policy.
+- Consumes batched events from `telemetry.batched.v1`.
+- Publishes/pins each batch artifact to IPFS (optionally XZ-compressed) and captures the CID.
+- Deduplicates by `batch_id` to avoid duplicate publication of redelivered batches.
+- Emits `ipfs.published.v1` with `cid` and `event_count`.
+- Commits `telemetry.batched.v1` offsets only after publish success and result emission.
 
 ### Robonomics Blockchain
 - Consumes IPFS-published events (`telemetry.ipfs.result.v1`) from Kafka.
@@ -81,6 +90,7 @@ The system accepts Ed25519-signed environmental sensor telemetry (Altruist-serie
 ## Core Kafka topics
 - `telemetry.authorized.v1`
 - `telemetry.rejected.v1`
+- `telemetry.batched.v1`
 - `ipfs.published.v1`
 - `telemetry.dlq.v1`
 
@@ -96,5 +106,6 @@ The system accepts Ed25519-signed environmental sensor telemetry (Altruist-serie
   - Endpoint -> PubSub
   - Endpoint -> IPFS
   - PubSub -> IPFS
+  - Batcher -> IPFS (must flow through Kafka)
   - IPFS -> Blockchain (must flow through Kafka)
 - Authentication is pluggable via `SensorAuth` interface but must use Redis for low-latency lookups.

docs/wp/wp-04-ipfs-publisher.md

@@ -1,7 +1,9 @@
 # WP-04 — `ipfs-publisher`
 
 ## Summary
-WP-04 implements deterministic batching and IPFS publication for authorized telemetry events. It consumes `telemetry.authorized.v1`, builds stable batch artifacts, publishes/pins to IPFS, and emits CID result events back to Kafka. The module is responsible for reliable side-effect sequencing with idempotent batch handling.
+WP-04 implements deterministic batching and IPFS publication for authorized telemetry events.
+
+> **Update (batcher/publisher split):** batching and IPFS publication are now separate services. The `@scp/batcher` service consumes `telemetry.authorized.v1`, builds deterministic batches, and emits `telemetry.batched.v1`. The `@scp/ipfs-publisher` service consumes `telemetry.batched.v1`, publishes/pins the batch artifact to IPFS, and emits `ipfs.published.v1` (`cid`, `event_count`). The sections below describe the combined responsibility; the batching tasks now belong to the batcher and the publication tasks to the ipfs-publisher.
 
 ## Depends on
 - WP-00

packages/core/src/topics.ts

@@ -16,6 +16,7 @@
 export const TELEMETRY_TOPICS = Object.freeze({
   AUTHORIZED: 'telemetry.authorized.v1',
   REJECTED: 'telemetry.rejected.v1',
+  BATCHED: 'telemetry.batched.v1',
   IPFS_PUBLISHED: 'ipfs.published.v1',
   DLQ: 'telemetry.dlq.v1',
 } as const);

packages/core/tests/core.test.ts

@@ -49,6 +49,7 @@ describe('core', () => {
   it('exposes exact stable topic constants', () => {
     expect(TELEMETRY_TOPICS.AUTHORIZED).toBe('telemetry.authorized.v1');
     expect(TELEMETRY_TOPICS.REJECTED).toBe('telemetry.rejected.v1');
+    expect(TELEMETRY_TOPICS.BATCHED).toBe('telemetry.batched.v1');
     expect(TELEMETRY_TOPICS.IPFS_PUBLISHED).toBe('ipfs.published.v1');
     expect(TELEMETRY_TOPICS.DLQ).toBe('telemetry.dlq.v1');
     expect(Object.isFrozen(TELEMETRY_TOPICS)).toBe(true);

proto/connectivity/v1/payload.proto

@@ -23,6 +23,22 @@ message TelemetryRejectedPayload {
   string reason_message = 3;
 }
 
+// TelemetryBatchedPayload carries a batch of authorized telemetry ready for
+// publication. It is emitted by the batcher and consumed by the ipfs-publisher.
+message TelemetryBatchedPayload {
+  // Unique batch identifier
+  string batch_id = 1;
+
+  // Serialized crypto.v1.SignedEnvelopeBatch (uncompressed)
+  bytes signed_envelope_batch = 2;
+
+  // Number of events in the batch
+  uint32 event_count = 3;
+
+  // Sensor IDs present in the batch (observability)
+  repeated bytes sensor_ids = 4;
+}
+
 // TelemetryIpfsPublishedPayload is emitted when events are published to IPFS.
 message TelemetryIpfsPublishedPayload {
   enum Compression {

services/batcher/README.md

@@ -0,0 +1,99 @@
+# `@scp/batcher`
+
+Groups authorized telemetry into deterministic batches and emits them for
+downstream publication.
+
+## Architecture
+
+- **Pattern**: Standard consumer with manual commit
+- **Input**: `telemetry.authorized.v1` from Kafka
+- **Output**: `telemetry.batched.v1` (carries a serialized `SignedEnvelopeBatch`)
+- **Autocommit**: Disabled (commit only after the batch is durably produced)
+- **Concurrency**: Single active flush per instance (single-flight); a timer-triggered flush cannot publish the same batch as a size/lag-triggered flush
+- **Shutdown**: Waits for any in-flight flush and flushes the remaining batch before closing resources
+
+Separating batching from IPFS publishing keeps the publisher stateless and
+lets each side scale and fail independently. The batcher owns buffering,
+flush timing, and offset commits for the authorized topic; the
+`@scp/ipfs-publisher` owns durable IPFS publication of `telemetry.batched.v1`.
+
+## Batching Strategy
+
+- **Size-based flush**: Flushes when `BATCHER_BATCH_SIZE` messages are accumulated
+- **Lag-based flush**: When consumer lag is high (>= batch size), flushes early to catch up
+- **Time-based flush**: When lag is low, a bounded timer (`BATCHER_BATCH_TIMEOUT_MS`) flushes partial batches
+- **Shutdown flush**: Any pending batch is flushed during graceful shutdown to avoid replay
+
+The current batch is atomically detached before any asynchronous work, so
+concurrent flush paths can never publish the same batch twice.
+
+## Environment
+
+- `KAFKA_BROKERS` (default: `localhost:9092`)
+- `BATCHER_GROUP_ID` (default: `batcher-v1`)
+- `BATCHER_SOURCE` (default: `batcher`)
+- `BATCHER_HEALTH_PORT` (default: `3041`)
+- `BATCHER_BATCH_SIZE` (default: `10`) - Maximum messages per batch
+- `BATCHER_BATCH_TIMEOUT_MS` (default: `30000`) - Timeout for partial batches under low load
+
+## Metrics
+
+Available at `http://localhost:3041/metrics`:
+
+- `consumed`: Total authorized telemetry messages consumed
+- `batchesProduced`: Total batches produced to `telemetry.batched.v1`
+- `eventsBatched`: Total individual telemetry events batched
+- `produceFailure`: Failed batch produce attempts (re-attached for retry)
+
+## Data Flow
+
+1. Consume `telemetry.authorized.v1` event from Kafka
+2. Extract `SignedEnvelope` from payload and add to the current batch
+3. When the batch is full, lag is high, or the flush timer fires:
+   - Serialize the batch as a `SignedEnvelopeBatch` protobuf
+   - Wrap it in a `TelemetryBatchedPayload` (`batch_id`, `event_count`, `sensor_ids`)
+   - Produce a `telemetry.batched.v1` envelope to Kafka
+   - Commit `telemetry.authorized.v1` offsets (only after the batch is produced)
+
+## Result Event Schema
+
+Published to `telemetry.batched.v1`:
+
+```typescript
+{
+  batchId: string;               // Unique batch identifier
+  signedEnvelopeBatch: Uint8Array; // Serialized crypto.v1.SignedEnvelopeBatch
+  eventCount: number;            // Number of telemetry events in batch
+  sensorIds: Uint8Array[];       // Sensor IDs present in the batch (observability)
+}
+```
+
+## Development
+
+```bash
+# Start batcher
+pnpm --filter @scp/batcher dev
+
+# Build
+pnpm --filter @scp/batcher build
+
+# Run tests
+pnpm --filter @scp/batcher test
+
+# Type check
+pnpm --filter @scp/batcher typecheck
+
+# Lint
+pnpm --filter @scp/batcher lint
+```
+
+## Dependencies
+
+- **Kafka**: For consuming authorized telemetry and producing batched events
+
+## Commit Safety
+
+The batcher commits `telemetry.authorized.v1` offsets only after the batch is
+durably produced to `telemetry.batched.v1`. If producing fails, the batch is
+re-attached and retried on the next flush, and offsets stay uncommitted so no
+telemetry is lost.

services/batcher/package.json

@@ -0,0 +1,26 @@
+{
+  "name": "@scp/batcher",
+  "version": "0.1.0",
+  "private": true,
+  "type": "module",
+  "main": "dist/index.mjs",
+  "types": "dist/index.d.mts",
+  "scripts": {
+    "build": "tsdown",
+    "typecheck": "tsc -p tsconfig.json --noEmit",
+    "lint": "eslint \"src/**/*.ts\"",
+    "test": "vitest run",
+    "dev": "tsx watch --env-file=../../.env src/index.ts"
+  },
+  "dependencies": {
+    "@buf/airalab_connectivity-protocol.bufbuild_es": "2.14.0-20260828134001-309d77314e7b.1",
+    "@platformatic/kafka": "^2.9.0",
+    "@scp/core": "workspace:*",
+    "pino": "^10.3.1"
+  },
+  "devDependencies": {
+    "@bufbuild/protobuf": "^2.14.0",
+    "tsdown": "^0.22.14",
+    "tsx": "^4.23.12"
+  }
+}