Repository navigation
feat: let an application wait for, or be notified when, the asynchronous work a command started has completed (fixes #711) - #723
Merged
Conversation
… into an application-registered recorder Spans, their kind, status and events live in Ecotone\Api\Tracing with the W3C traceparent and tracestate headers as the only carrier; the tracer, interceptors and module live in Ecotone\Tracing and register only when a TracingConfiguration names a SpanRecorder or a tracing package hands over a Tracer, so an application without tracing pays nothing. The suite mirrors the OpenTelemetry package's tree, correlation and asynchronous scenarios against the in-memory recorder.
…og lines become span events, and failed sends leave no span open Edge cases proved against the in-memory recorder: a failing or filtering channel interceptor, a message stripped of its carrier, an invalid traceparent, a failing error channel, a failing distributed bus send, and two tracers refused at bootstrap.
…d of producing spans itself OpenTelemetryTracer starts live SDK spans with the ids Core reports and hands every ended span to the configured recorders; the module registers it through a TracerReference only when a TracerProviderInterface was given, so a flow test without one traces through Core's own tracer. TracerInterceptor, TracingChannelInterceptor, AddSpanEventLogger and the JSON carrier header are gone, and TracingConfiguration moved to Ecotone\Api\OpenTelemetry\OpenTelemetryConfiguration. The 41 package tests pass unchanged apart from that rename.
… is a consumer span, and spans carry the messaging semantic conventions The event bus route resolver is a replaceable service the tracing module wraps, so each route resolved for an event becomes a 'Routing to: <endpointId>' or 'Routing to Projection: <name>' span between the Event Bus and the handler, synchronous or asynchronous, and a failing synchronous handler fails its route. The projection trigger endpoint carries ProjectionTriggerEndpoint, which the tracing interceptor turns into a 'Projection: <name>' consumer span. Every span names messaging.system, its operation type, its destination and its endpoint or projection through SpanAttribute.
…dlers completed, and notify subscribers after commit Enterprise. TracingConfiguration::withStoredTraces() turns every span into a stored one: a unit of work buffers its spans and writes them once, inside the transaction, before commit; the root span is written when it ends, after commit, and only then are #[OnTraceCompleted] and #[OnSpanEnded] subscribers notified. The TraceCompletion gateway reads the ledger derived from the spans: a trace is complete when its root ended and every route it sent has an ended handler span, a destination is complete when its handler ended, and await() polls until then or fails naming what is still pending. SpanContext moved to Api so an application can mint the trace id it later asks about. Ships with the in-memory store; the DBAL store follows.
…ritten inside the handler's transaction Enterprise. DbalTraceStore claims a row per trace, locks it while it appends the unit of work's spans and decides completion from the same ledger the in-memory store uses, and marks the trace completed exactly once so subscribers are notified once across concurrent consumers. The tables come from ecotone:migration:database:setup, a flow test without the connection falls back to the in-memory store, and ecotone:tracing:remove-expired-traces applies the configured retention. One conformance suite runs both stores, plus the cross-instance visibility case only the database holds.
…ndler without an explicit endpointId still completes its route The route span and the handler span used to read the endpoint id from two attribute instances, which generate different ids when none is declared, so such a trace never completed. The route now writes the destination into the W3C tracestate header, which travels with the message through synchronous and asynchronous channels, and the handler span names that destination. The upgrade guide gains the stored traces entry and the planned depth completion and opt-in recording.
…umed, its root is the span that started it, a failed unit of work discards its spans, and a subscriber's bus calls do not notify it again An asynchronous command handler, or any message sent to a channel outside an event route, completed its trace the moment the bus returned: only event routes recorded a destination. The ledger now also pairs every successful channel send with the consumer span that processed the same message id, and names the channel when it is pending. Any span whose parent was not stored counted as the ended root, so a consumer that stored its spans before, or instead of, its producer completed the trace on its own. The span started outside any other span, unless it is a consumer, now carries ecotone.trace.root and only that ends the root. A failed unit of work kept its buffered spans and wrote them with the failed outer span, so a handler that succeeded before the rollback counted as completed and a send the rollback removed counted as pending forever. The buffer is discarded on failure. Publication ran inside the span end and traced the subscriber's own bus calls, so a subscriber that sent a command completed a new trace and was notified again without end. Notifications gathered while publishing are dropped.
…resOutsideTransaction opts in to the write after the rollback The failed outer span was always written after the rollback, the one stored-trace write that happens outside a transaction. It is now dropped by default, and TracingConfiguration::withStoredTraces(storeFailuresOutsideTransaction: true) restores it. The flow-testing fallback that moves stored traces in memory carries the flag through.
…d without a licence, and the refusal names what to remove Only withStoredTraces() without a licence was tested; #[OnTraceCompleted] and #[OnSpanEnded] reach the same guard but had no test of their own, and the message named neither the configuration that triggered it nor the way out besides buying a licence. It now lists the usages found and says to remove them to run without one.
…ention, {operation} {target}, with the handler's qualified name in code.function.name
The 1.x names were sentences, Sending to Channel: orders, Command Handler: App\OrderService::place, inconsistent in
case and long. They are now send command, send event, send query, send distributed, route <endpointId>,
send <channel>, receive <channel>, process <Class::method> and process <projection>, so a trace UI groups them as it
groups any other messaging span. The fully qualified handler stays on the span as code.function.name.
…confirm # Conflicts: # packages/Ecotone/Api/Messaging/MessageHeaders.php # packages/Ecotone/src/Messaging/Support/FrameworkHeaders.php # upgrade/namespace-map-2.0.csv
…ers use the ecotone_{module}_{feature}_{action} convention
ServiceActivatorBuilder became MessageProcessorActivatorBuilder::createForMethod() and DbalConnectionFactory moved to
Api on main. The gateway headers of TraceCompletion are ecotone_tracing_completion_destination and
ecotone_tracing_completion_timeout; traceparent and tracestate keep their W3C names because they leave the process.
…side by side across an asynchronous DBAL channel A cross-module flow test enables the DBAL and tracing packages together with a real database and an in-memory OpenTelemetry exporter: one traceparent is exported by the SDK, stored in ecotone_traces and ecotone_spans, kept pending until the DBAL channel's consumer ran, and notified once; a second application on the same database sees the completion.
…or channel or the acknowledgement strategy is stored as failed, not as processed The consumer span wraps the error-channel and acknowledgement interceptors, which swallow the handler's exception when the consumer does not stop on error, so the span ended Ok and paired the delivery as processed while the message was requeued or dead-lettered. The recorder now remembers which open span wrapped a discarded unit of work and stores that span as failed, unless a later retry in the same span flushed successfully.
… claims completion after the commit, without a row lock SELECT ... FOR UPDATE on the trace row was a pessimistic lock, which rule 8 forbids, and it still could not protect a decision made from a transaction's own snapshot. The flush inside the transaction now only appends. The write of the span wrapping the unit of work, which follows the commit, reads what every unit of work has committed by then and claims completion with one conditional update, so exactly one of two concurrent finishers claims it and neither can miss the other's commit.
…d a handler's destination comes from its registration instead of tracestate A flow entered through a #[MessageGateway] or #[OrchestratorGateway] had no span around it, so every step of an orchestrator started its own trace and a stored trace had no unit of work to buffer in. Both are now traced as send <requestChannel> or send orchestrator and open the recording scope, like a bus. The route stamped its destination into tracestate so that a handler without an explicit endpointId could still name it, but a reply copies the request headers, so the handler behind an outputChannelName hop was counted under the previous destination and one delivery completed the destination of two. The module now reads every handler's registered endpoint id at compile time, the same attribute instances the routing module used, and the interceptor looks its own up by class and method. tracestate is no longer written. Flow shapes now covered: outputChannelName chains, splitters, routers, orchestrators entered through their gateway and reached from a command handler, message gateways, and a command bus inside a command handler.
…rted as attributes, framework attribute names cannot be forged, and stored traces without a store are refused at bootstrap Every scalar user header became a span attribute, which stored tokens and personal data for the retention period and let a header named ecotone.endpoint.id or ecotone.trace.root pose as a framework attribute. The attribute builder is now a service configured with the headers to report, defaulting to all, and never reports a header under the ecotone., messaging. or code. prefixes. withStoredTraces() with no TraceStoreReference and inMemory false used to fall back to a per-process store in production; it is refused with both ways out. The OpenTelemetry package has a test that it traces nothing until a TracerProviderInterface is registered.
…es its asynchronous target, the handler span is innermost, and the timeout names failures, unknown traces and unreached destinations A send to an asynchronous handler reached the ledger with no destination unless it came from an event route, so two commands to one handler completed it after the first. The channel send span now carries the target endpoint id read from the routing slip, and a destination tracks the message ids dispatched to it, so a redelivered or replayed message satisfies its own delivery only. A channel send outside any span continues the message's own traceparent, so a dead-letter replay completes the trace it came from. The handler span was the outermost interceptor and ended Ok when an error channel inside it handled the failure; it is now the innermost, at Precedence::TRACING_HANDLER_PRECEDENCE, so the handler's own failure is stored inside the transaction that moved its message. With storeFailuresOutsideTransaction the failed unit's spans are kept. The recording interceptor has its own Precedence::TRACE_RECORDING_PRECEDENCE, the following constants moved up by one, ecotone_spans is keyed by (trace_id, span_id), and TraceCompletion is registered unconditionally, refusing with the configuration it needs when stored traces are off.
… trace completes again, retention spares pending traces, appends are idempotent, polling projections are refused, and a handler can read its trace id A trace that was handed more work after completing, as when one request sends two commands under one traceparent, reopens on append and notifies again when it completes again. The store records when a completion was notified; a subscriber that throws no longer fails the committed command, and any later unit of work re-delivers a notification left undelivered for thirty seconds. Span inserts are idempotent and the buffer is taken before the write, so a failed write cannot replay the same span ids. The retention command deletes completed traces past the retention and any trace past twice the retention with two statements, instead of loading every id into one IN list and deleting traces whose delayed messages are still queued. isCompletedFor() and awaitFor() refuse a #[Polling] projection with the reason. Ecotone\Api\Tracing\CurrentTrace gives a handler the id of the trace it runs in, and refuses with the way to enable tracing when nothing records one.
… the outbox and dead-letter replay, a no-op OpenTelemetry provider gets its own span ids, and multi-tenant stored traces use the default connection The first channel a message passes stamps ecotone_tracing_delivery with its name and the message id, so the inner channel of a dynamic channel and a redelivery do not count as deliveries nobody consumes, and the consumer pairs the message with the channel it was first sent to. A high-throughput batch envelope lists the deliveries of its entries, and every entry receives the envelope's traceparent, so the consumer continues the producer's trace instead of starting its own. An OpenTelemetry span that is not recording, as the no-op provider returns, no longer lends its ancestor's ids to every stored span. A multi-tenant application stores traces on the configured default connection, since the write after the unit of work has no tenant, and is refused at bootstrap when there is none.
…stores only requested traces, tracestate travels with the trace, and a flush inserts its spans in one statement A trace whose root continued a traceparent is marked carried, the mark travels across channels and the distributed bus in the W3C tracestate as ecotone=carried next to any other vendor's entry, and the recorder ignores every other span, so ordinary traffic costs no storage. SpanContext carries the trace state, so a vendor entry received with a command reaches the asynchronous handler. The DBAL store inserts all spans of a flush in one statement.
… handing the core module their recorder through a module extension TracingModule stays Apache-2.0 and knows nothing of stored traces beyond composing the recorders it is handed: the licence guard, the subscriber guard, the store, the recording interceptor, the completion gateway, the retention command and the polling-projection refusal moved to StoredTracesModule, which announces its recorder with a SpanRecorderReference. The completion gateway registration both modules need is a shared helper.
… of work, so the recorder keeps no bookkeeping of foreign spans and the failure shows in OpenTelemetry too The storing recorder remembered the ids of spans wrapping a discarded unit of work and rewrote their status when they arrived, state that outlived the unit of work and was invisible to the OpenTelemetry export. The handler span now marks the active span that wraps it as failed when the handler throws and as recovered when a later attempt succeeds, and that span ends as failed on its own, in both tracers. The recorder's publication queues are gone; it keeps only the spans of the current unit of work until the write that follows the commit.
…nsaction, so an outage after a commit cannot fail a consumer or a command The span wrapping a unit of work was inserted when it ended, after the commit, together with the completion claim, and with storeFailuresOutsideTransaction the failed spans were inserted after the rollback. With the database down at that moment the write threw out of a consumer whose message was already acknowledged, or out of a command whose data had committed, instead of leaving the message to the retry. Now the flush inside the transaction writes the buffered spans together with the wrapping span as it stands at that point and claims completion in the same transaction; an interceptor at the new Precedence::TRACE_PUBLICATION_PRECEDENCE, outside the instant retries and the transaction, publishes the committed spans and completions from memory and never touches the database. Queries are no longer recorded. storeFailuresOutsideTransaction, the notified_at column and the post-commit notification retry are removed; notifications are at most once, and failures are visible in OpenTelemetry. AGENTS.md and the coding conventions gain rule 16a: no database write after the transaction, telemetry excepted.
…oneLite, and rule 10 says so explicitly TraceStoreContractTest constructed InMemoryTraceStore and DbalTraceStore and called them; the two behaviours it guarded, completing once and reopening after new work, are already proved through the bus and the TraceCompletion gateway, so it is removed. The outage tests dropped Ecotone's tables; they now hand Ecotone a Doctrine connection whose statements fail on demand, switched by an #[Around] interceptor next to the transaction, and a third test proves an outage inside the consumer's transaction leaves the message to the instant retry. Edge cases that used internal channel classes moved to the DBAL package on that connection: a send to a channel, to the error channel and through the distributed bus failing with the database, and a consumer receiving a message without or with an invalid traceparent from an application that does not trace. A failure before the asynchronous send is a #[Presend] interceptor. A message dropped by a channel interceptor, and two tracers registered at once, cannot happen from an application: the first test is gone, and so is the two-tracer guard with its test. AGENTS.md rule 10 and the conventions now say it in so many words: no class outside Api/ in a test, not even to prove two implementations alike, no table touched to simulate a failure, and code no application can reach is removed rather than tested from the inside.
…letion is renamed TracingGateway An application names the flow it wants to track with MessageHeaders::TRACING_ID, a UUID or 32 hexadecimal characters, and passes the same value to TracingGateway, whose methods and TraceCompleted now speak of tracingId. The first span turns it into the W3C traceparent, so a message that already carries a traceparent from an instrumented request keeps that trace; carrying both is refused, and so is a tracingId in neither format, each naming the way out. tracingId is a framework header, not propagated to child messages, and a consumer continues the traceparent its channel stamped. CurrentTrace::traceId() becomes tracingId().
…ns live in the DBAL package under FlowTrackingConfiguration TracingGateway, #[OnTraceCompleted], #[OnSpanEnded] and what they need only work with storage, so they moved with the store into Ecotone\Api\Dbal\Tracing and one DbalTracingModule. FlowTrackingConfiguration::createWithDefaults() tracks every flow with the gateway and the notifications on; withTracingGateway(false) and withNotifications(false) turn each off while the flows are still tracked, and an attribute left behind with notifications off is refused at bootstrap. withRetention() and withOnlyRequestedFlows() replace the withStoredTraces() arguments, and flow testing switches withInMemoryStorage() on when no connection is given. Core keeps spans, recorders, CurrentTrace and the OpenTelemetry seam, and no longer names a DBAL class; a subscriber's own handler is traced like any other.
…mpleted] and FlowCompleted, matching flow tracking
…pCompleted::$step, and TraceCompletionTimedOut is FlowTrackingTimedOut A subscriber only ever receives steps whose unit of work committed, so the notification says a step of the flow completed; the step is still the OpenTelemetry span.
…its own, so a subscriber can never reopen or re-notify the flow it was told about A subscriber's work, synchronous or sent through an asynchronous channel, joined the flow it was notified about: the consumer re-completed the flow and notified again, forever. A nested unit of work also reset the publishing guard mid-delivery. The notification now starts a root span marked ecotone=untracked in tracestate, nothing carrying that mark is stored, and a subscriber failure is logged instead of vanishing.
…h transaction, never after it committed Ecotone turns the endpoint transaction off for an asynchronous projection, which commits once per batch, so flow tracking wrote the projection's spans after the read model had committed - a database outage at that moment failed the consumer for work that succeeded. Flow tracking now joins each batch's transaction and marks the projection processed in the batch that finds it caught up. Paths the application runs with no transaction at all still write when their work ends.
…s database, inside the tenant's transaction Flow tracking wrote on the default connection while the work committed on the tenant's, so a flow could read as completed before the tenant's data committed, and stayed completed when that commit failed. Spans are now written on the tenant's connection, TracingGateway takes the tenant as metadata (refusing a call without one, naming it), and the retention command runs per tenant with --header.
…eam of it can still send it another message isCompletedFor() and awaitFor() returned as soon as everything sent to the destination so far was processed, so a later message the same flow was still going to send it was not waited for. A pending message now blocks a destination unless the destination's own handling caused it, followed through each span's parentId.
…ndler is refused, naming the flow the call belongs to A nested call's tracingId was silently ignored: the call joined the outer flow, and awaiting the new id timed out with a message that still said traceparent. The bus and gateway entries now refuse it, and the timeout messages speak of flows and tracingId.
…mer of an untraced message complete The query and distributed buses were not recorded, and a consumer starting a fresh trace never marked its root, so such flows stayed pending forever and their timeout listed nothing. Both buses are now recorded, a consumer without a remote parent is the root of its flow, and a tracingId that names the flow the call already runs in - as a distributed message carries across services - is no longer refused.
… id its route was compiled with The handler span looked its endpoint id up by the runtime class, so a container that serves a subclass proxy of the handler missed it, the span got a fresh id and its route stayed pending forever. The lookup now walks up to the declaring class.
… notification is refused at bootstrap A #[OnFlowCompleted] or #[OnFlowStepCompleted] method declaring the wrong type failed on every notification, at runtime, after the work committed. The module now checks the first parameter against FlowCompleted or FlowStepCompleted when it prepares, naming the type to declare.
…guration registered twice are refused at bootstrap #[Asynchronous] on a flow subscriber asked for an endpointId the attribute cannot take; an endpoint attribute can now state why it cannot run asynchronously, and AsynchronousModule refuses with that reason. A second FlowTrackingConfiguration was dropped silently by the flow-testing rewrite; it is now refused, naming how to register one.
… other usage is refused rather than given a meaning of its own
…hort transaction ecotone:tracing:remove-expired-traces deleted every expired flow in two statements, so the first run after a long gap held one large delete. It now removes batches of 1000 trace ids - their spans and flow rows together - until none is left; FlowTrackingConfiguration::withRemovalBatchSize() changes the size and refuses one below 1. The upgrade guide now says the command has to be scheduled.
…n or every handler in it isCompletedFor() and awaitFor() take a class as the destination: a #[Projection] class resolves to its projection name, any other class to the endpoint ids of every handler method in it, explicit or generated, mapped at compile time. The class completes once every one of its handlers the flow reached ran and nothing upstream can still send it more. A class Ecotone registered no handler or projection for is refused at once instead of timing out.
…ointing to flow notifications An await blocks the calling process while it polls. A timeout above 15 minutes is now refused before any polling, naming #[OnFlowCompleted] and #[OnFlowStepCompleted] so a long flow is reacted to when it completes instead of waited for.
…eturning a boolean, and fetchFlow() returning a Flow isCompletedFor() takes an optional destination and replaces isCompleted(); isCompletedWithin() replaces await() and awaitFor(), returning false once the timeout passed instead of throwing FlowTrackingTimedOut, which is removed. fetchFlow() returns a Flow - isCompleted, pending, reached and its stored steps in start order - or null for a flow nobody recorded, so a false is explained by reading the flow.
… the form main now requires
…oduleClassList, as it does every other DBAL module
…ut depending on a 1 ms poller firing twice in one run The test relied on a fixed-rate poller executing twice inside one run(), which the CI runners did not do; its lenient tree comparison then matched one span against both expected nodes. Two runs of one execution each now produce the two polls, and the test asserts both spans, both roots, on different traces.
…ace with its handler beneath it, without depending on poller timing Like the scheduled-handler test, it relied on a 1 ms poller firing twice in one run(), and its lenient tree comparison could match both expected runs against one. Two runs of one execution each now produce the two polls, and the test asserts both root spans, each with its own add handler span, on different traces.
…its completion once on MySQL and MariaDB too On REPEATABLE READ the second consumer waited for the first one's commit, then read the flow's spans from its earlier snapshot, missed the first part, and neither part claimed completion, so #[OnFlowCompleted] never fired. The flow row now carries a version: a part writes it only if no other part committed since it read it, and otherwise rolls back with ConcurrencyException so the retry reads the flow from a fresh snapshot. A forked-process race test proves one notification on PostgreSQL, MySQL and MariaDB.
…er, because completion is read from the steps instead of claimed on a shared flow row #[OnFlowCompleted] needed one part of a flow to claim that it completed it, which made every part write one shared row per flow and, under contention, rolled one part back and ran its handler twice. Completion was always decided from the stored steps, so it is now only read: isCompletedFor() and fetchFlow() answer from ecotone_tracing alone, each step is its own row, and two consumers of one flow commit side by side. - #[OnFlowCompleted], FlowCompleted, the ecotone_traces table and its version are removed; a #[OnFlowStepCompleted] subscriber asks TracingGateway::isCompletedFor() after its commit, so the part that commits last always sees the flow complete - retention keeps every flow for withRetention() from its first step, removed from the spans in batches
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Why is this change proposed?
Why
Important
Flow tracking (Enterprise,
ecotone/dbal) adds one table,ecotone_tracing: runecotone:migration:database:setup, and scheduleecotone:tracing:remove-expired-traces, which nothing runs for you.An application can now ask whether everything a command triggered has finished — asynchronous handlers and projections included — wait for it with a timeout, or be notified when it completes. Until now a command returned once its synchronous part was done, so a UI or an API had to poll a read model and guess when a projection had caught up (#711). Completion is decided from Ecotone's own tracing, recorded in OpenTelemetry vocabulary, with OpenTelemetry as one exporter of it.
Resulting flow
Before
sequenceDiagram App->>CommandBus: PlaceOrder CommandBus-->>App: returns (async work still queued) App->>ReadModel: poll until it looks updatedAfter
sequenceDiagram App->>CommandBus: PlaceOrder + tracingId CommandBus-->>App: returns Consumer->>Flow tracking: handler/projection committed App->>TracingGateway: isCompletedWithin(tracingId, 5s, OrdersList::class) TracingGateway-->>App: true Flow tracking-->>Subscriber: #[OnFlowStepCompleted] FlowStepCompletedDescription of Changes
Turning it on
Every flow is tracked and kept for 7 days, with the gateway and notifications on. Flow tracking needs an Enterprise licence and
ecotone/dbal.Waiting for a flow —
TracingGatewayNaming a flow with
tracingId. The application chooses a UUID (or 32 hex characters) and sends it with the command, event, query or gateway call. Use case: a controller returns the id to the client, which asks about it later.Waiting until it completes —
isCompletedWithin(). Polls for at most the timeout, then returnstrueorfalse. Use case: read-your-writes — after a form submit, wait until the order is in the list, then redirect to it.Use case: an end-to-end test waits for asynchronous processing instead of sleeping.
Asking once —
isCompletedFor(). Answers without waiting. Use case: a status endpoint the frontend polls.Waiting for one part of the flow — destinations. A destination is an
endpointId, a projection name, or a class: a#[Projection]class stands for its projection, any other class for every handler in it. Use case: the page shows the order list, so wait for that projection, not for the email and invoice the same command also triggered.Seeing what happened —
fetchFlow(). Returns aFlow—isCompleted,pendinghandlers and channels,reacheddestinations, and every stored step — ornullfor a flow nobody recorded. Use case: explain afalsein a log or to support: which consumer has not run yet, which handler failed.Long waits are refused. A timeout above 15 minutes is refused, pointing to
#[OnFlowStepCompleted], because a wait holds the calling process for its whole length.Reacting to a flow — notifications
#[OnFlowStepCompleted]receives aFlowStepCompletedfor every committed step, with its span. Use case: a live progress view for a long import.Reacting to completion. The subscriber asks
isCompletedFor()for the whole flow or one destination. It runs after its own commit, so the part of the flow that commits last always sees the flow complete. Use case: tell the browser over a websocket that the order list is ready.When two parts of a flow commit at the same moment, both can see it complete, so a reaction to completion is at-least-once.
A subscriber runs synchronously after the commit. Work meant for the background is a command it sends to an
#[Asynchronous]handler, and that work is never tracked itself, so a subscriber cannot re-trigger the flow it was told about.Tuning flow tracking —
FlowTrackingConfigurationwithOnlyRequestedFlows(true)stores only flows sent with atracingIdortraceparent. Use case: in production, only the requests a UI waits on cost anything.withNotifications(false)/withTracingGateway(false)keep tracking but turn one side off. Use case: an application that only reacts to completion, or only waits for it.withRetention()/withRemovalBatchSize()set how long a flow is kept from its first step, completed or not, and how many expired flows the retention command removes per transaction. Use case: keep flows for a day, cleaned up hourly from cron.Multi-tenant. Each flow is stored in its tenant's database, inside the tenant's transaction. The gateway takes the tenant as metadata, and the retention command runs per tenant.
Across services. A
tracingIdsent through the distributed bus follows the message to the receiving service. With both services sharing the tracking database, the sender can wait for the receiver.In tests. Without a database connection, flow tracking runs in memory, with the same behaviour.
Tracing (open source, core)
Spans cover every bus call, gateway and orchestrator call, handler, route, projection, distributed send, and channel send and receive. They are named the way OpenTelemetry expects, and joined into one trace across channels and services through the W3C
traceparent/tracestateheaders. Use case: follow one HTTP request through every asynchronous consumer in Jaeger.OpenTelemetry is enabled by registering a
TracerProviderInterface. A trace started by an instrumented HTTP request is continued.withSpanAttributesFromHeaders()chooses which message headers become span attributes; by default every application header does. Use case: search Jaeger by tenant or order, without leaking other headers.withSpanRecorder()andInMemorySpanRecorderrecord spans without a collector. Use case: assert in a test which handlers a command reached.CurrentTrace::tracingId()gives a handler the id of the flow it runs in. Use case: write it to logs, so a log line leads to its flow.Guarantees, and where to look
withTransactionOnCommandBus(false)— writes when its work ends.parentId. A message a handler forwards unchanged through an output channel is waited for too: it has the same message id as a sibling's copy of an event, so the two cannot be told apart.#[Asynchronous], aFlowTrackingConfigurationregistered twice, subscribers with notifications off. Wrong input is refused when the call is made: a nested or doubledtracingId, a polling projection or a class without handlers as a destination, a wait over 15 minutes, a multi-tenant call without its tenant.Where to start reading
packages/Dbal/src/Tracing/TraceRecordingInterceptor.php— when spans are stored and notifications delivered.packages/Dbal/src/Tracing/TraceProgress.php— when a flow or a destination counts as complete.packages/Ecotone/src/Tracing/TracingInterceptor.php— where spans start and howtracingIdmaps to a trace.AGENTS.mdgains rules 16a and 1d and an explicit userland-only test rule;upgrade-2.0.md§14 documents the span names and flow tracking.Pull Request Contribution Terms