Concurrency View

Describes the concurrent and parallel execution units in the system: processes, threads, background services, async I/O, and the coordination mechanisms used to prevent race conditions.

Process & Thread Model

The entire platform runs in a single ASP.NET Core process hosted by Kestrel. HTTP and WebSocket connections share the Kestrel thread pool; each request flows asynchronously through the MediatR pipeline. Three IHostedService background services run on the same process concurrently: the outbox dispatcher (domain event delivery), the vase health monitor (IoT liveness), and the MQTT startup subscriber (IoT message ingestion). No inter-process communication is needed within the API boundary.

flowchart TD
  subgraph APIProcess["ASP.NET Core API Process (Kestrel)"]
    TP["Kestrel Thread Pool\n(handles HTTP + WebSocket requests)"]
    MQ["MediatR Pipeline\n(per-request async chain)"]
    SIG["SignalR Hub\n(BouquetHub — in-process)"]

    subgraph BGServices["Background Services (IHostedService)"]
      OBD["OutboxDispatcherService\nPolls DB every 5s"]
      VHM["VaseHealthMonitoringService\nPolls DB every 60s for stale heartbeats"]
      MQS["MqttDeviceStartupService\nSubscribes to MQTT topics on startup"]
    end
  end

  TP --> MQ
  TP --> SIG
  MQ --> BGServices

Sub-domain Mapping — Process Model

Execution Unit	Business Sub-domain	Role
Kestrel thread pool (HTTP)	All sub-domains	Handles every inbound REST request — vendor management, customer purchasing, analytics, admin
Kestrel thread pool (WebSocket)	IoT / Real-time	Maintains persistent SignalR connections for live bouquet freshness and order status updates
MediatR pipeline	All sub-domains	Routes commands and queries through cross-cutting behaviours (logging, validation, tenant check, idempotency, cache)
OutboxDispatcherService	IoT / Real-time + Customer Purchasing	Delivers domain events (order confirmed, bouquet updated) from the DB outbox to the in-process event bus every 5 s
VaseHealthMonitoringService	IoT / MQTT	Marks vases Offline when no heartbeat has been received for 1 hour; publishes VaseWentOfflineEvent
MqttDeviceStartupService	IoT / MQTT	Subscribes to MQTT topics (vases/+/heartbeat, vases/+/sensor) on API startup; feeds telemetry into the system

Concurrency Hotspots

1. Bouquet Reservation Race — Customer Purchasing / Vase & Bouquet Lifecycle

Multiple customers may attempt to purchase the same bouquet simultaneously. The reservation is enforced via a pessimistic DB-level check inside the CreateOrderCommandHandler: bouquet status is read and updated in a single transaction, and the EF Core concurrency token prevents double-reservation.

sequenceDiagram
  participant C1 as Customer A
  participant C2 as Customer B
  participant H as CreateOrderCommandHandler
  participant DB as PostgreSQL (transaction)

  par Concurrent requests
    C1->>H: CreateOrderCommand(bouquetId)
    C2->>H: CreateOrderCommand(bouquetId)
  end

  H->>DB: BEGIN TX\nSELECT ... FOR UPDATE (bouquet row)
  Note over DB: Only one TX holds the lock
  DB-->>H: bouquet.Status = Available → set Reserved
  H->>DB: COMMIT (Order created, BouquetReservedUntil = now+15min)

  DB-->>H: Second TX: bouquet.Status = Reserved → reject
  H-->>C2: 409 Conflict (bouquet-unavailable)
  H-->>C1: 201 Created

2. Cart Vendor-Mismatch Guard — Customer Purchasing

Cart is read-modify-write; the vendor-scope check is done inside the Cart aggregate method AddItem(), which is protected by the database transaction in the command handler. No distributed lock is needed since a customer has exactly one cart row (keyed by CustomerId).

3. Outbox Dispatcher — At-Least-Once Delivery — All Sub-domains (event backbone)

sequenceDiagram
  participant OBD as OutboxDispatcherService
  participant DB as PostgreSQL
  participant BUS as InMemoryEventBus

  loop Every 5s
    OBD->>DB: SELECT top 20 WHERE processed_at IS NULL ORDER BY created_at
    DB-->>OBD: [OutboxMessage]
    OBD->>BUS: PublishAsync(event)
    Note over OBD,BUS: If crash here, message replays on next poll (at-least-once)
    OBD->>DB: UPDATE SET processed_at = now()
  end

Known gap (EP-09): RabbitMQEventBus.PublishAsync is a stub — events are silently dropped when RabbitMQ bus is active. InMemoryEventBus is the real implementation in current use.

4. Vase Health Monitor — IoT / MQTT

sequenceDiagram
  participant VHM as VaseHealthMonitoringService
  participant DB as PostgreSQL
  participant BUS as InMemoryEventBus

  loop Every 60s
    VHM->>DB: SELECT SmartVases WHERE LastHeartbeat < now()-1hr AND Status != Offline
    DB-->>VHM: [stale vases]
    VHM->>DB: UPDATE ConnectionStatus = Offline
    VHM->>BUS: Publish VaseWentOfflineEvent
  end

5. Idempotency — Duplicate Request Deduplication — Customer Purchasing / Payments

Redis SET NX (via IdempotencyMiddleware) serialises concurrent duplicate requests at the middleware layer before they reach MediatR, preventing duplicate order creation even if the client retries concurrently.

sequenceDiagram
  participant C1 as Client (retry 1)
  participant C2 as Client (retry 2)
  participant MW as IdempotencyMiddleware
  participant Redis

  par
    C1->>MW: POST /orders  Idempotency-Key: abc
    C2->>MW: POST /orders  Idempotency-Key: abc
  end
  MW->>Redis: SET idempotency:abc NX EX 86400
  Note over Redis: Only one SET wins
  Redis-->>MW: OK (first) / nil (second)
  MW-->>C2: 201 (replayed from cache — no second order)

Coordination Mechanisms Summary

Mechanism	Scope	Used For
PostgreSQL row-level lock (FOR UPDATE)	Per-aggregate	Bouquet reservation, order creation
EF Core concurrency token	Per-entity	Optimistic concurrency on critical entities
Redis SET NX	Per idempotency key	Duplicate HTTP request deduplication
Outbox pattern	Cross-service	Exactly-once domain event emission guarantee
Single background thread per service	Per IHostedService	Outbox poll, vase health poll (no inter-thread sharing)
SignalR hub groups	Per WebSocket session	Scoped real-time broadcast (no shared mutable state)

Cache Invalidation Concurrency

Cache entries use a version key strategy: each write command increments a Redis counter {entity-type}:version; readers embed the version in the cache key. If a write races with a read, the version mismatch causes a cache miss on the next read cycle — there is no window for stale reads beyond one cache-TTL period.