From 7aa717f746e13ecfb8e4852c404253dcf6552eac Mon Sep 17 00:00:00 2001 From: Dominic Finn Date: Sun, 4 Oct 2026 19:04:35 +0100 Subject: [PATCH] Deliver events to handlers in every deployment (#327) PgBossEventBus only fanned events out to handlers registered in its own process, so with DISABLE_EMBEDDED_WORKERS=true every event the API published was recorded and then delivered nowhere. A process that runs no handlers now puts events on an evt.__dispatch queue, and every process that has started handlers consumes it and fans them out to its handler queues. This also covers WORKER_MODE=integrations workers. The standalone worker now sets up the same DI container as the API (queue adapter, event bus, full command bus, delivery and arrival services) instead of hand-built command buses, and both processes register the event handlers with the full command bus. The command-driven handlers (issue engine, auto-replenishment, route planning) therefore run in both setups, whether or not an LLM is configured. Replaces the route-planning-only workaround from #336. Co-Authored-By: Claude Opus 5.5 --- .../__tests__/events/PgBossEventBus.test.ts | 60 ++++++++++++++++ backend/src/events/PgBossEventBus.ts | 52 +++++++++++--- backend/src/events/registerHandlers.ts | 11 --- backend/src/index.ts | 11 +-- backend/src/worker.ts | 72 ++++++------------- docs/DOMAIN_BEHAVIOURS.md | 22 ++++-- docs/SELF_HOSTING.md | 3 +- docs/roadmap/completed.md | 1 + 8 files changed, 148 insertions(+), 84 deletions(-) create mode 100644 backend/src/__tests__/events/PgBossEventBus.test.ts diff --git a/backend/src/__tests__/events/PgBossEventBus.test.ts b/backend/src/__tests__/events/PgBossEventBus.test.ts new file mode 100644 index 00000000..98dd4003 --- /dev/null +++ b/backend/src/__tests__/events/PgBossEventBus.test.ts @@ -0,0 +1,60 @@ +import { PgBossEventBus } from '../../events/PgBossEventBus'; +import { createTestEvent } from '../helpers/testUtils'; + +function setup() { + const subscriptions = new Map Promise>(); + const queue = { + publish: jest.fn().mockResolvedValue('job-1'), + subscribe: jest.fn(async (name: string, handler: any) => { subscriptions.set(name, handler); }), + } as any; + const prisma = { domainEventLog: { create: jest.fn().mockResolvedValue({}) } } as any; + const bus = new PgBossEventBus(prisma, queue); + const published = () => queue.publish.mock.calls.map((c: any[]) => c[0]); + return { bus, queue, subscriptions, published }; +} + +const event = () => createTestEvent('shipment.created', 'shipment', 'ship-1', {}); + +describe('PgBossEventBus fan-out (#327)', () => { + it('hands events to the dispatch queue in a process that does not run handlers', async () => { + const { bus, published } = setup(); + // Routing registered but never started: the API with DISABLE_EMBEDDED_WORKERS=true. + await bus.subscribe('projection.shipment', ['shipment.*'], jest.fn()); + + await bus.publish(event()); + + expect(published()).toEqual(['evt.__dispatch']); + }); + + it('fans out directly to matching handler queues in a process that runs handlers', async () => { + const { bus, published, subscriptions } = setup(); + await bus.subscribe('projection.shipment', ['shipment.*'], jest.fn()); + await bus.subscribe('projection.order', ['order.*'], jest.fn()); + await bus.start(); + + await bus.publish(event()); + + expect(published()).toEqual(['evt.projection.shipment']); + expect([...subscriptions.keys()]).toEqual(['evt.projection.shipment', 'evt.projection.order', 'evt.__dispatch']); + }); + + it('fans dispatched events out to its own handler queues', async () => { + const { bus, published, subscriptions } = setup(); + await bus.subscribe('audit', ['*'], jest.fn()); + await bus.start(); + + await subscriptions.get('evt.__dispatch')!({ type: 'shipment.created', payload: event() }); + + expect(published()).toEqual(['evt.audit']); + }); + + it('keeps dispatching when started with no handlers, so nothing is dropped', async () => { + const { bus, published, subscriptions } = setup(); + await bus.start(); + + await bus.publish(event()); + + expect(published()).toEqual(['evt.__dispatch']); + expect(subscriptions.has('evt.__dispatch')).toBe(false); + }); +}); diff --git a/backend/src/events/PgBossEventBus.ts b/backend/src/events/PgBossEventBus.ts index 6490d2c8..676cc020 100644 --- a/backend/src/events/PgBossEventBus.ts +++ b/backend/src/events/PgBossEventBus.ts @@ -19,6 +19,14 @@ import { IQueueAdapter, QueueMessage } from '../queue/IQueueAdapter.js'; const QUEUE_PREFIX = 'evt.'; +/** + * Events published by a process that doesn't run handlers (the API with DISABLE_EMBEDDED_WORKERS, + * or an integrations-only worker) can't be fanned out there: fan-out only knows the handlers + * registered in its own process (#327). They go to this queue instead, and whichever process does + * run handlers consumes it and fans them out to its handler queues. + */ +const DISPATCH_QUEUE = QUEUE_PREFIX + '__dispatch'; + interface HandlerRegistration { patterns: string[]; handler: EventHandler; @@ -31,6 +39,8 @@ type TxClient = Omit(); private started = false; + /** True once start() wired at least one handler: this process consumes events itself. */ + private consuming = false; constructor( private prisma: PrismaClient, @@ -61,22 +71,26 @@ export class PgBossEventBus implements IEventBus { * Fan out an already-persisted event to matching handler queues. * Called after transaction commit. Failures are logged but don't throw * because the event is already safely in DomainEventLog. + * + * A process that runs handlers fans out itself; one that doesn't hands the event to the dispatch + * queue, for a process that does (#327). */ async fanOut(event: DomainEvent): Promise { + if (!this.consuming) { + await this.queue.publish(DISPATCH_QUEUE, toMessage(event)).catch((err) => { + console.error(`[EventBus] Dispatch failed for event ${event.id} (${event.type}): ${(err as Error).message}`); + }); + return; + } + await this.fanOutLocally(event); + } + + private async fanOutLocally(event: DomainEvent): Promise { const fanOutPromises: Promise[] = []; for (const [handlerName, registration] of this.registry) { if (this.matchesAny(event.type, registration.patterns)) { - const queueName = QUEUE_PREFIX + handlerName; - const message: QueueMessage = { - type: event.type, - payload: event, - metadata: { - timestamp: event.timestamp, - sourceId: event.id, - }, - }; - fanOutPromises.push(this.queue.publish(queueName, message)); + fanOutPromises.push(this.queue.publish(QUEUE_PREFIX + handlerName, toMessage(event))); } } @@ -141,11 +155,20 @@ export class PgBossEventBus implements IEventBus { await this.wireHandler(handlerName, reg.handler, reg.options); } this.started = true; + + // A process with handlers also takes events dispatched by processes without any. + if (this.registry.size > 0) { + await this.queue.subscribe(DISPATCH_QUEUE, async (message: QueueMessage) => { + await this.fanOutLocally(message.payload as DomainEvent); + }); + this.consuming = true; + } console.log(`[EventBus] Started with ${this.registry.size} handler(s)`); } async stop(): Promise { this.started = false; + this.consuming = false; console.log('[EventBus] Stopped'); } @@ -190,3 +213,12 @@ export class PgBossEventBus implements IEventBus { }); } } + +function toMessage(event: DomainEvent): QueueMessage { + return { + type: event.type, + payload: event, + metadata: { timestamp: event.timestamp, sourceId: event.id }, + }; +} + diff --git a/backend/src/events/registerHandlers.ts b/backend/src/events/registerHandlers.ts index 799529d6..edd4d5a8 100644 --- a/backend/src/events/registerHandlers.ts +++ b/backend/src/events/registerHandlers.ts @@ -250,14 +250,3 @@ function shipmentRoutePlanningHandler(prisma: PrismaClient, commandBus: ICommand )); } -/** - * Route planning for the embedded (API process) workers, which register the other handlers without - * a command bus (#328). The standalone worker gets it through registerEventHandlers instead, so it - * is never registered twice. - */ -export async function registerShipmentRoutePlanning(eventBus: IEventBus, prisma: PrismaClient, commandBus: ICommandBus): Promise { - const handler = shipmentRoutePlanningHandler(prisma, commandBus); - await eventBus.subscribe(handler.name, handler.eventPatterns, (event) => handler.handle(event), handler.options); - console.log(`[EventBus] Registered handler: ${handler.name} (patterns: ${handler.eventPatterns.join(', ')})`); -} - diff --git a/backend/src/index.ts b/backend/src/index.ts index db7e4fde..18487f8a 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -24,7 +24,7 @@ import { createOrderAutoArchiveWorker, registerOrderAutoArchiveSchedule, ORDER_A import { OrderAutoArchiveService } from './services/OrderAutoArchiveService.js'; import { createCarrierUserAnonymizeWorker, registerCarrierUserAnonymizeSchedule, CARRIER_USER_ANONYMIZE_QUEUE } from './workers/carrierUserAnonymizeWorker.js'; import { CarrierUserAnonymizationService } from './services/CarrierUserAnonymizationService.js'; -import { registerEventHandlers, registerShipmentRoutePlanning } from './events/registerHandlers.js'; +import { registerEventHandlers } from './events/registerHandlers.js'; import { ICommandBus } from './commands/CommandBus.js'; import type { IEventBus } from './events/IEventBus.js'; import { createWebhookRetryWorker, registerWebhookRetrySchedule, WEBHOOK_RETRY_QUEUE } from './workers/webhookRetryWorker.js'; @@ -153,16 +153,19 @@ async function start() { if (process.env.DISABLE_EMBEDDED_WORKERS !== 'true') { // CQRS event handlers (projections, audit, notifications) — wired in the // backend process for dev. In prod with a separate worker container, - // DISABLE_EMBEDDED_WORKERS=true keeps these out so they aren't processed twice. + // DISABLE_EMBEDDED_WORKERS=true keeps these out so they aren't processed twice; this process + // then runs no handlers, so its events go to the dispatch queue for the worker (#327). try { const eventBus = container.resolve(TOKENS.IEventBus); const documentService = container.has(TOKENS.IDocumentGenerationService) ? container.resolve(TOKENS.IDocumentGenerationService) : undefined; + // With the command bus, the command-driven handlers (issue engine, auto-replenishment, + // route planning) run here too, as they do in the standalone worker (#327). await registerEventHandlers( - eventBus, server.prisma, undefined, undefined, undefined, undefined, undefined, documentService + eventBus, server.prisma, undefined, undefined, undefined, + container.resolve(TOKENS.ICommandBus), undefined, documentService ); - await registerShipmentRoutePlanning(eventBus, server.prisma, container.resolve(TOKENS.ICommandBus)); await eventBus.start(); server.log.info('Embedded event handlers registered (projections + audit + notifications)'); } catch (err) { diff --git a/backend/src/worker.ts b/backend/src/worker.ts index 375a7476..aab83e43 100644 --- a/backend/src/worker.ts +++ b/backend/src/worker.ts @@ -4,7 +4,7 @@ * This process: * - Does NOT start Fastify or listen on any HTTP port * - Creates its own PrismaClient with its own connection pool - * - Creates its own PgBossQueueAdapter + * - Sets up the same DI container as the API (queue adapter, event bus, full command bus) * - Registers event handlers (audit, notifications, email, webhooks, triage) * - Optionally runs the existing operational workers (inbound webhook) * @@ -20,17 +20,10 @@ */ import { PrismaClient } from '@prisma/client'; -import { PgBossQueueAdapter } from './queue/PgBossQueueAdapter.js'; import { PgBossEventBus } from './events/PgBossEventBus.js'; import { registerEventHandlers } from './events/registerHandlers.js'; import { QUEUES } from './queue/events.js'; import { createInboundWebhookWorker } from './workers/inboundWebhookWorker.js'; -import { OrderDeliveryService } from './services/OrderDeliveryService.js'; -import { ArrivalCriteriaEvaluationService } from './services/ArrivalCriteriaEvaluationService.js'; -import { RecordGeofenceArrivalCommandHandler } from './commands/tracking/RecordGeofenceArrivalCommand.js'; -import { RecordGeofenceDepartureCommandHandler } from './commands/tracking/RecordGeofenceDepartureCommand.js'; -import { RecordJourneyCheckpointCommandHandler } from './commands/tracking/RecordJourneyCheckpointCommand.js'; -import { RecordStopOrdersDeliveryCommandHandler } from './commands/orders/RecordStopOrdersDeliveryCommand.js'; import { IEmailService } from './services/IEmailService.js'; import { SmtpEmailService } from './services/SmtpEmailService.js'; import { ConsoleEmailService } from './services/ConsoleEmailService.js'; @@ -39,14 +32,13 @@ import { DatabaseBinaryStorage } from './storage/DatabaseBinaryStorage.js'; import { S3FileStorage } from './storage/S3FileStorage.js'; import { AnthropicLlmProvider } from './services/llm/AnthropicLlmProvider.js'; import { ILlmProvider } from './services/llm/ILlmProvider.js'; -import { CommandBus, ICommandBus } from './commands/CommandBus.js'; -import { CreateAgentDecisionCommandHandler } from './commands/agentDecisions/CreateAgentDecisionCommand.js'; -import { RecordDecisionOutcomeCommandHandler } from './commands/agentDecisions/RecordDecisionOutcomeCommand.js'; -import { PromoteDecisionCommandHandler } from './commands/agentDecisions/PromoteDecisionCommand.js'; -import { CreateIssueCommandHandler } from './commands/issues/CreateIssueCommand.js'; -import { UpdateIssueCommandHandler } from './commands/issues/UpdateIssueCommand.js'; -import { EscalateIssueCommandHandler } from './commands/issues/EscalateIssueCommand.js'; -import { SetShipmentRouteCommandHandler } from './commands/shipments/SetShipmentRouteCommand.js'; +import { ICommandBus } from './commands/CommandBus.js'; +import { registerDependencies } from './di/index.js'; +import { container } from './di/container.js'; +import { TOKENS } from './di/tokens.js'; +import type { IQueueAdapter } from './queue/IQueueAdapter.js'; +import type { IOrderDeliveryService } from './services/OrderDeliveryService.js'; +import type { IArrivalCriteriaEvaluationService } from './services/ArrivalCriteriaEvaluationService.js'; import { DEFAULT_TRIAGE_PROMPT, DEFAULT_TRIAGE_EVENTS } from './events/handlers/TriageAgentHandler.js'; import { SkillRegistry } from './services/skills/SkillRegistry.js'; import { DocumentGenerationService, IDocumentGenerationService } from './services/DocumentGenerationService.js'; @@ -103,15 +95,14 @@ async function startWorker() { await prisma.$connect(); console.log('[Worker] Database connected'); - // Own queue adapter - const dbUrl = process.env.DATABASE_URL || ''; - const queue = new PgBossQueueAdapter(dbUrl); + // The same container as the API (#327): one queue adapter, one event bus and the full command + // bus, so every command-driven handler runs here whether or not an LLM is configured. + registerDependencies(prisma); + const queue = container.resolve(TOKENS.IQueueAdapter); await queue.start(); console.log('[Worker] Queue adapter started'); - - // One event bus for the process. Fan-out only reaches handlers registered on this bus, so the - // integration workers below publish through the same instance the event handlers register on. - const eventBus = new PgBossEventBus(prisma, queue); + const eventBus = container.resolve(TOKENS.IEventBus); + const commandBus = container.resolve(TOKENS.ICommandBus); // Event handlers (audit, notifications, email, webhooks, triage) if (WORKER_MODE === 'all' || WORKER_MODE === 'events') { @@ -156,7 +147,6 @@ async function startWorker() { // LLM key may only drive it when that org is the only tenant. With several orgs, one tenant's // key would pay for, and see, every other tenant's agent traffic, so the environment is used. let llmProvider: ILlmProvider | undefined; - let workerCommandBus: ICommandBus | undefined; const soleOrgId = await resolveSoleOrganizationId(prisma); const org = soleOrgId @@ -177,17 +167,6 @@ async function startWorker() { baseURL: process.env.ANTHROPIC_BASE_URL, }); - // Worker-local command bus for agent handlers to dispatch commands - const bus = new CommandBus(); - bus.register(new CreateAgentDecisionCommandHandler(prisma, eventBus)); - bus.register(new RecordDecisionOutcomeCommandHandler(prisma, eventBus)); - bus.register(new PromoteDecisionCommandHandler(prisma, eventBus)); - bus.register(new CreateIssueCommandHandler(prisma, eventBus)); - bus.register(new UpdateIssueCommandHandler(prisma, eventBus)); - bus.register(new EscalateIssueCommandHandler(prisma, eventBus)); - // The route planning handler stores routes through this bus (#328). - bus.register(new SetShipmentRouteCommandHandler(prisma, eventBus)); - workerCommandBus = bus; const source = org?.llmApiKey ? 'org config' : 'env var'; console.log(`[Worker] LLM provider configured (Anthropic via ${source}), AI agents enabled`); @@ -199,10 +178,8 @@ async function startWorker() { // Build skill registry for automation rules (always available, even without LLM) const skillRegistry = new SkillRegistry(); - if (workerCommandBus) { - skillRegistry.register(new CreateIssueSkill(workerCommandBus)); - skillRegistry.register(new EscalateIssueSkill(workerCommandBus)); - } + skillRegistry.register(new CreateIssueSkill(commandBus)); + skillRegistry.register(new EscalateIssueSkill(commandBus)); skillRegistry.register(new CallWebhookSkill()); if (emailService) { skillRegistry.register(new SendEmailSkill(emailService)); @@ -221,7 +198,7 @@ async function startWorker() { ) : undefined; - await registerEventHandlers(eventBus, prisma, emailService, storageProvider, llmProvider, workerCommandBus, skillRegistry, documentService); + await registerEventHandlers(eventBus, prisma, emailService, storageProvider, llmProvider, commandBus, skillRegistry, documentService); await eventBus.start(); console.log('[Worker] Event handlers registered and started'); } @@ -229,18 +206,11 @@ async function startWorker() { // Integration workers (inbound webhook). The legacy outbound carrier and // outbound tracking workers were removed — outbound EDI is now driven by // Edi856AutoSendHandler and Edi810AutoSendHandler off domain events. - // Registered after the event handlers so the bus already knows where to fan out (#287). + // In integrations-only mode this process runs no handlers, so its events go to the dispatch queue + // for an events worker to fan out (#327). if (WORKER_MODE === 'all' || WORKER_MODE === 'integrations') { - if (WORKER_MODE === 'integrations') { - console.warn('[Worker] integrations-only mode: tracking events are recorded but only fan out to handlers registered in this process'); - } - const trackingCommandBus = new CommandBus(); - trackingCommandBus.register(new RecordGeofenceArrivalCommandHandler(prisma, eventBus)); - trackingCommandBus.register(new RecordGeofenceDepartureCommandHandler(prisma, eventBus)); - trackingCommandBus.register(new RecordJourneyCheckpointCommandHandler(prisma, eventBus)); - trackingCommandBus.register(new RecordStopOrdersDeliveryCommandHandler(prisma, eventBus)); - const deliveryService = new OrderDeliveryService(prisma, trackingCommandBus); - const arrivalCriteriaService = new ArrivalCriteriaEvaluationService(prisma, deliveryService, trackingCommandBus); + const deliveryService = container.resolve(TOKENS.IOrderDeliveryService); + const arrivalCriteriaService = container.resolve(TOKENS.IArrivalCriteriaEvaluationService); await queue.subscribe( QUEUES.INBOUND_WEBHOOK, createInboundWebhookWorker(prisma, deliveryService, arrivalCriteriaService, eventBus), diff --git a/docs/DOMAIN_BEHAVIOURS.md b/docs/DOMAIN_BEHAVIOURS.md index 0ed1da65..1d9f9494 100644 --- a/docs/DOMAIN_BEHAVIOURS.md +++ b/docs/DOMAIN_BEHAVIOURS.md @@ -25,6 +25,16 @@ API Request → Validate (Zod) → Dispatch Command → Execute in Transaction Every event is persisted to the immutable `DomainEventLog` before handler fan-out. Events can be replayed, exported for data warehouses, or consumed by ML pipelines. +**Fan-out across processes (#327).** `PgBossEventBus` fans an event out to the queues of the handlers +registered in its own process. A process that runs no handlers (the API with +`DISABLE_EMBEDDED_WORKERS=true`, or a `WORKER_MODE=integrations` worker) can't do that, so it puts +the event on one `evt.__dispatch` queue instead; every process that has started handlers also +consumes that queue and fans each event out to its own handler queues. pg-boss delivers each job +once, so nothing is handled twice, and scaled-out workers share the dispatch queue. Both the API's +embedded workers and the standalone worker register the event handlers with the full command bus, +so the command-driven handlers (issue engine, auto-replenishment, route planning) run in either +setup; the triage agent still also needs an LLM provider. + --- ## Orders @@ -473,13 +483,11 @@ its readings only in the webhook log's raw payload. The webhook log's response b `readingsReceived` and `readingsStored`. System Loco readings are still stored by `SystemLocoAdapter` (it feeds cold chain and alerting). -**Standalone worker (#287).** `worker.ts` (the separate worker container) now wires the same journey -pipeline as the embedded API worker: it builds one `PgBossEventBus` for the process, registers the -event handlers on it first, then builds a command bus with the three tracking command handlers and an -`ArrivalCriteriaEvaluationService`, and passes both the service and the bus into -`createInboundWebhookWorker` (the worker has no DI container to resolve them from). Fan-out only -reaches handlers registered in the publishing process, so in `WORKER_MODE=integrations` tracking -events are recorded in the event log but reach no handlers. +**Standalone worker (#287, #327).** `worker.ts` (the separate worker container) sets up the same DI +container as the API, so it uses the same queue adapter, event bus, full command bus, order delivery +service and arrival-criteria service, and wires the same journey pipeline as the embedded API worker. +In `WORKER_MODE=integrations` it runs no event handlers, so its events go through the dispatch queue +to an events worker. **Skipped checkpoints.** Pings can be far enough apart that a shipment jumps several checkpoints. `RecordJourneyCheckpointCommand` records every checkpoint between the last recorded one and the one diff --git a/docs/SELF_HOSTING.md b/docs/SELF_HOSTING.md index 815a8a69..379c0e99 100644 --- a/docs/SELF_HOSTING.md +++ b/docs/SELF_HOSTING.md @@ -162,7 +162,8 @@ to start. Setting `DISABLE_EMBEDDED_WORKERS=true` moves that work to a separate `dist/worker.js` process, so the API only serves requests. That split is worth it once request latency and background work start -competing for CPU. Check the open issues first: the separate worker process has a known problem that +competing for CPU. Events the API publishes then go to an `evt.__dispatch` queue, which the +worker consumes and fans out to its handlers. Check the open issues first: the separate worker process has a known problem that makes it unrunnable at present. ### EDI collector diff --git a/docs/roadmap/completed.md b/docs/roadmap/completed.md index ca727318..a2212d93 100644 --- a/docs/roadmap/completed.md +++ b/docs/roadmap/completed.md @@ -55,6 +55,7 @@ and priorities). Newest entries append at the bottom of their phase; new phases - **Shipment Routes (#328, part 1)** - Shipments on a custom route (no lane) get their own planned road route, calculated through their stops in order when they're created or changed while draft or ready, and only when the stops change. Checkpoints, route deviation alerts and the shipment map use the shipment's own route, or its lane's. - **Create Shipments from Orders (#328, part 2)** - "Create shipment" on an order and "Ship together" on the orders list open the create page with the orders attached in an Orders card, with an "Add order" search for compatible orders. The orders fill in the form without locking anything, a matching lane is preselected, and the page refuses to save a shipment the orders can't share. - **Changing Shipments Built from Orders (#328, part 3)** - Editing a draft shipment keeps its stops, so its orders keep their drop stops (they used to be cut on every edit). Adding an order puts its drop before the destination and re-plans a custom route. Orders combined or converted through the API land on a lane only when it matches their stops exactly. A lane shipment whose stops no longer match its lane shows a warning with "Switch to custom route" or "Save as new lane". +- **Events Reach Handlers in Every Deployment (#327)** - With the API's embedded workers turned off, events it published used to be recorded and then delivered to nothing, so projections, notifications and webhooks never ran. A process that runs no handlers now hands events to a dispatch queue that the worker fans out. The standalone worker uses the same DI container as the API, and both register the command-driven handlers (issue engine, auto-replenishment, route planning) with the full command bus, whether or not an LLM is configured. - **Exceptions** - Exception status with type classification, resolution workflow, event-driven notifications, ETA-based auto-detection - **Normalised Tracking Pings (#323)** - Every IoT payload (System Loco or the generic webhook) is parsed into one internal ping: device timestamp, position, a readings array (temperature, battery, light, pressure) and a readings count. Generic pings now store their telemetry against the registered device (idempotent on retry) and drive arrivals, origin departure and journey checkpoints, which previously only System Loco did. Arrival, departure, checkpoint and order-delivery times use the device's timestamp instead of processing time. Checkpoints skipped between sparse pings are filled in on the planned route with interpolated times. - **Multi-Stop Geofence Journeys (#324)** - Every stop on a shipment now completes from geofence events, not just the origin and destination: entering a delivery stop's geofence completes it and delivers its orders, in any order. The shipment only completes once every stop is done; reaching the final stop with stops unvisited raises a `stops_not_visited` exception. Repeat visits to the same location are handled as separate stops. Departing the origin moves a ready shipment to in progress, checkpoints exclude the origin and destination geofences, events carry the device id, and the timeline shows origin arrival (#307 items 1-3).