Broker / Message Bus — транспортный слой, через который producers публикуют сообщения, а один или несколько consumers получают их по routing rules, topics, subscriptions или consumer groups без прямой жёсткой связи producer → конкретный process.
№42 Events & Triggers владеет domain/event semantics: что произошло и какая reaction возможна; №59 переносит сообщение. №57 Queues & Workers владеет executable jobs, leases, attempts и worker lifecycle; broker может доставлять jobs, но это отдельная responsibility. №58 Scheduler создаёт time-originated occurrence; broker лишь транспортирует его. №50 Contracts задаёт message schemas/version compatibility. №61 Provenance/Lineage отслеживает origin/transformations data; broker хранит message metadata/correlation, но не весь lineage. №68 Durable Workflow владеет process history/replay; broker replay не равен workflow replay.
Prerequisites: №42 Events, №46 Observability, №50 Contracts, №57 Queues & Workers, №58 Scheduler. Forward references: №61 Provenance, №63 Retry/Circuit Breakers, №64 Rate Limits/Budgets, №68 Durable Workflow, №69 Distributed Reliability.
REQUEST-TIME: producer may publish synchronously, but consumers usually async. CONTROL PLANE: topics, subscriptions, routing, retention, schemas, ACLs, partitions. DATA PLANE: message envelopes and delivery/ack state. OFFLINE: replay, re-drive, schema migration tests, load/failure tests.
Success: accepted message reaches all required durable subscriptions according to delivery contract. Retryable: transient broker/consumer/network outage. Permanent: invalid schema, forbidden topic, unsupported version. Delivery: duplicates expected unless effect-level guarantees prove otherwise. Idempotency: message_id/event_id + consumer-side dedupe/effect key. Persist: message metadata, topic, partition/routing key, schema version, offsets/ack state where relevant. Trace: publish→broker→subscription→consume→ack.
№59 не владеет domain event meaning, task execution lifecycle, timers, workflow state, data warehouse, source ingestion semantics или business compensation. Она владеет MESSAGE ROUTING, DELIVERY, FAN-OUT AND SUBSCRIPTION TRANSPORT.
Service A вызывает B, C и D напрямую.
Добавление E требует менять A; failure одного consumer влияет на producer path.
A публикует stable event/message в topic.
Broker хранит routing/delivery semantics.
B/C/D/E подписываются самостоятельно и развиваются независимо.
Одна logical job должна быть выполнена одним из workers. Несколько workers конкурируют за work item.
Пример: document.parse.
Одно событие может получить каждый заинтересованный consumer/subscription.
Пример: document.updated → search index + audit + analytics.
{
"message_id": "MSG-...",
"message_type": "document.updated",
"schema_version": "2.0",
"occurred_at": "...",
"published_at": "...",
"tenant_id": "tenant_A",
"producer": "ingestion-service",
"subject_ref": "doc://tenant_A/123",
"correlation_id": "TRACE-...",
"causation_id": "MSG-parent-...",
"routing_key": "tenant_A.document",
"payload": {
"source_version": "v42"
}
}| Type | Meaning | Typical ownership |
|---|---|---|
| EVENT | Fact: something already happened. | Producer owns fact; consumers choose reaction. |
| COMMAND | Request that a specific capability perform an action. | Usually one logical handler / work queue. |
| NOTIFICATION | Informational signal, may be lossy depending contract. | Consumer may ignore. |
| STATE SNAPSHOT | Current representation of object/state. | Useful for projection/cache/update, but may be large/stale. |
send_email событием, если это command. Точная семантика сильно влияет на retry, idempotency и ownership.document.events, workflow.events, audit.events.
document.updated, document.deleted.
tenant/resource/category/region where topology requires.
Messages with same key land in same ordered partition if broker supports it.
Subscription progress retained. Consumer after restart continues from last acknowledged offset/message.
Подходит для index, audit, business projections.
Old messages may be irrelevant. Подходит для live UI hints, transient monitoring, noncritical notifications.
Hours/days/weeks based on recovery, audit and governance needs.
Новый/починенный consumer может пройти historical events заново.
Replay public/purchase/send commands without idempotency may repeat real-world actions.
| Ordering scope | Example | Recommendation |
|---|---|---|
| None | Independent analytics events. | Max throughput; consumer handles concurrency. |
| Per entity | Updates for one document/customer/workflow. | Partition key = entity id. |
| Per tenant | Tenant-local ordered projection. | Can create hotspot for large tenants. |
| Global | Total system sequence. | Avoid unless domain truly requires it. |
Event carries source/entity version so projection can ignore stale update.
Useful but wall-clock alone is weaker than authoritative sequence/version.
For critical mutable state, event can trigger live read of current object.
Consumers ignore unknown optional fields.
Keep old field/schema until all consumers migrate.
Use explicit schema version and compatibility strategy.
Schema registry becomes useful at scale; simple repo contracts are enough initially.
Application updates DB, then publishes event. Crash between operations leaves state changed without event — or event published before DB commit.
Business state and outbox row commit together. Separate relay publishes outbox messages and marks them delivered/retries idempotently.
BEGIN; UPDATE documents SET version = 42 WHERE id = 123; INSERT INTO outbox( message_id, topic, payload, status ) VALUES (..., 'document.events', ..., 'PENDING'); COMMIT; OUTBOX RELAY: claim pending row publish to broker mark SENT / retry CONSUMER: dedupe by message_id / effect key
consumer(message):
begin transaction
if inbox.exists(
consumer="search-indexer",
message_id=message.id
):
return ACK
apply_projection(message)
inbox.insert(
consumer="search-indexer",
message_id=message.id
)
commit
ACKConsumer-side inbox/dedupe table помогает:
Unsupported schema, invalid payload, deterministic consumer bug.
After bounded attempts, message/subscription delivery moves to dead-letter destination/state.
After consumer/schema repair, controlled re-drive preserves original id/correlation.
Каждая durable subscription идёт своим темпом.
Consumer lag/oldest unprocessed age — primary health signal.
Добавлять consumers while preserving required ordering and downstream capacity.
Ephemeral/low-value consumers may drop or sample under pressure if contract allows.
Producer identity scoped to allowed topics/message types.
Consumer access restricted by topic/tenant/data classification.
Use references/capabilities; avoid raw credentials in messages.
Tenant metadata validated and applied to routing/storage/consumer authorization.
document.updated
correlation = TRACE-7
Indexer processes A and emits document.indexed.
correlation = TRACE-7
causation_id = MSG-A
Consumer события A пишет state, который снова публикует A без change detection.
Unbounded downstream fan-out exceeds quotas or creates cascading backlog.
Change detection, causation tracing, hop limits where relevant, dedupe, rate/admission controls.
Messages retained to recover consumers, replay processing or audit transport.
Events are canonical source of truth and current state is derived by replay. Requires stronger domain/version/invariant discipline.
Services independently react to facts. Хорошо, когда no single component owns full end-to-end process.
Нужны explicit state, steps, timers, retries, compensation, completion criteria, history/replay.
Messages/sec by topic/type/producer.
Producer → broker accepted.
Oldest unprocessed age / offset distance.
published_at → consumer process/ack.
Repeated message ids delivered/observed.
Count, age, types, ownership.
Messages produced per input event; detect storms.
Rejected schema/ACL/retention expiration where measurable.
Same message twice → one logical consumer effect.
Old entity version after new one does not corrupt projection.
Durable subscription catches up after restart.
Moves to dead-letter after bounded attempts, does not block forever.
Crash between DB write and publish is eventually reconciled.
Producer/outbox buffers safely, no silent loss.
Old/new producer-consumer versions coexist.
Unauthorized subscriber cannot read another tenant's protected stream.
Message throughput by topic/type/producer.
Oldest unprocessed age/offset distance per subscription.
published → consumer effect/ack.
Repeated message IDs and duplicate logical effects.
Poison/incompatible messages by contract and consumer.
Downstream messages per source message/event.
Oldest committed event not yet published.
Messages rejected for unsupported/invalid contract version.
messages/
├── contracts/
├── publisher.py
├── subscriptions.py
├── outbox.py
├── inbox.py
├── handlers/
└── tests/
outbox(
message_id,
topic,
message_type,
schema_version,
tenant_id,
payload_json,
status,
created_at,
published_at
)
consumer_inbox(
consumer_id,
message_id,
processed_at,
primary key(consumer_id, message_id)
)
MVP options:
A) PostgreSQL outbox + polling consumers
B) DB queue + fan-out tables
C) lightweight broker when multiple
independent consumers justify itДля небольшого монолита PostgreSQL outbox + background dispatcher может дать основную decoupling/reliability ценность без отдельного cluster.
| Signal | Why broker helps |
|---|---|
| Many independent consumers / fan-out | Subscriptions, consumer groups, routing become first-class. |
| High throughput / sustained message volume | Dedicated log/broker handles streaming better than OLTP DB tables. |
| Need retention/replay | Historical stream can rebuild projections/consumers. |
| Service boundaries across hosts/teams | Decoupled transport and ownership become valuable. |
| Partitioned ordered streams | Broker partitions/offsets support scalable ordered consumption. |
| Independent backpressure per consumer | Each subscription advances at its own rate. |
| Вопрос | Ответ |
|---|---|
| Стоит ли реализовывать? | Только когда есть реальная multi-consumer/event-driven topology. Для малого монолита direct calls + PostgreSQL queue/outbox часто лучше. |
| Separate Component? | YES как transport responsibility. Физически может появиться позже, когда dedicated broker оправдан. |
| Минимум 80% ценности? | Stable message contracts, outbox, idempotent consumers, topics/subscriptions, schema versioning, correlation, retries/DLQ, lag metrics. |
| Когда overkill? | Kafka cluster ради двух фоновых handlers и десятка событий в час. |
| Trigger? | One producer needs multiple independent consumers, replay/retention, decoupled services, high-volume asynchronous message routing. |
| Как измерить uplift? | Producer/consumer coupling, fan-out reliability, consumer lag, recovery/replay time, duplicate effects, incident isolation, time to add new consumer. |
| Можно ли rule/tool/code вместо LLM-agent? | Полностью. Broker/message bus — deterministic infrastructure. LLM may produce domain data, but never owns delivery semantics. |
Сначала решить event/command meaning, затем выбирать broker/topic.
message_id/schema_version обязательны для replay/dedupe/audit.
Consumer effects должны быть idempotent.
Prefer entity/partition ordering over global ordering.
DB state and publish intent commit together.
Slow consumer should not block unrelated consumers.
Replaying transport history must not repeat unsafe side effects.
Correlation + causation reveal event chains and storms.
Monolith + DB outbox is a valid intermediate architecture.
DOMAIN / SYSTEM COMPONENT
↓
SOMETHING HAPPENS
↓
CREATE VERSIONED MESSAGE
message_id
message_type
schema_version
tenant
subject
occurred_at
correlation
causation
↓
[ TRANSACTIONAL OUTBOX IF DB STATE CHANGED ]
↓
PUBLISH
↓
BROKER / MESSAGE BUS
topics
routing
partitions
retention
subscriptions
↓
├─ CONSUMER A
│ ↓
│ validate
│ idempotent effect
│ ACK / offset
│
├─ CONSUMER B
│ ↓
│ own independent lag/state
│
└─ CONSUMER C
↓
own independent lag/state
FAILURES:
duplicate → inbox/idempotency
stale/out-of-order → version check
poison → DLQ
consumer offline → durable subscription
producer crash → outbox replay
message storm → backpressure / quotas
BOUNDARIES:
№42 EVENT/TRIGGER
= WHAT HAPPENED / WHAT IT MEANS
№57 QUEUE/WORKER
= WHAT WORK IS WAITING / WHO EXECUTES
№58 SCHEDULER
= WHEN A TIME OCCURRENCE EXISTS
№59 BROKER
= HOW MESSAGES ARE ROUTED
AND DELIVERED TO CONSUMERS
CORE PRINCIPLE:
A MESSAGE BUS SHOULD
REDUCE COUPLING,
NOT HIDE OWNERSHIP.
IT MOVES VERSIONED,
TRACEABLE MESSAGES.
IT DOES NOT DECIDE
WHAT THE BUSINESS MEANS,
WHETHER A WORKFLOW IS COMPLETE,
OR WHETHER AN EXTERNAL EFFECT
IS SAFE TO REPEAT.
A. Related ECC ideas. Context-as-cache, scoped memory, lifecycle hooks, selective capabilities, feature flags, deterministic enforcement, provider-neutral adapters and eval-gated learning are applied only where relevant to №59 Broker / Message Bus.
B–E. Existing boundary and placement. The existing conceptual boundary, class PRODUCTION, default CONDITIONAL and owner Production Fabric remain authoritative. Runtime/control/data/offline placement is unchanged; durable state stays outside model context.
F–H. Hooks and contracts. Use bounded PRE_MODEL/POST_MODEL, PRE_TOOL/POST_TOOL, CHECKPOINT and TASK_COMPLETED events as applicable. Illustrative fields and canonical contracts are defined in NEW_CONTRACTS_SPEC.md; no universal schema is implied.
I–J. Security and evaluation. Host-side schema, permission, secret, budget, idempotency and audit checks take precedence over LLM output. Optional mechanisms require a feature flag and WITH/WITHOUT ablation; measure quality, acceptance, correction, latency, cost, escalations and severe errors.
K–L. Task profiles and cross-references. A TaskProfile selects the relevant skill, tool/context slice, memory scope and enforcement profile independently from FAST/STANDARD/DEEP. See cross-reference map, hook spec and ablation plan. Provider adapters remain outside the core.