--- title: "Async Engine (Go and C++)" description: "The Go SDK's asyncengine subpackage and the C++ SDK's openpit::asyncengine::TypedAsyncEngine turn an AccountSync engine into a concurrent facade with." --- # Async Engine (Go and C++) The Go SDK's `asyncengine` subpackage and the C++ SDK's `openpit::asyncengine::TypedAsyncEngine` turn an `AccountSync` engine into a concurrent facade with per-account ordering guarantees. These helpers are one possible integration pattern - applications may provide their own sharded queues, actor model, or equivalent dispatcher. Both SDKs provide sharded and dynamic strategies, futures, observer callbacks, graceful stop, and hard stop. The Go examples below use goroutines and `context.Context`; the C++ surface uses worker threads, `Future`, and `std::chrono` deadlines. The wrapped `AccountSync` engine receives at most one operation per account at a time, and queued operations for an account reach it in submission order. ## When to Use Use a bundled async engine when the application: - builds the engine with `AccountSync` to allow concurrent access across many accounts; - wants the SDK to guarantee that no two operations for the same account ever run inside the engine concurrently; - prefers `Future`-style return values over manual channel/wait-group bookkeeping; - needs graceful and hard stop modes with deadlines. If `FullSync` or `NoSync` is sufficient, or if you already have a custom per-account dispatcher you are happy with, you do not need this package. ## Choosing Between Strategies `asyncengine` ships two dispatch strategies. Each chosen at build time so the hot path stays branch-free in steady state. ### Sharded - Go: `asyncengine.NewBuilder(engine).Sharded(workers)` - C++: `openpit::asyncengine::MakeTypedAsyncEngine(engine, workers)` or `TypedBuilder(driver).Sharded(workers)` The two C++ forms differ only in who owns the engine adapter. The typed builder borrows a driver that the caller keeps alive, which is what a custom driver needs. `MakeTypedAsyncEngine` is the shortcut for the common case of wrapping a plain engine: it owns the default adapter itself and exposes the typed async methods - start pre-trade, execute pre-trade, apply execution report, apply account adjustment, and apply drop copy - without any driver lifetime plumbing. It does not own the source engine. The same engine object must outlive the async wrapper at its original address and must not be moved while the wrapper exists; rvalue engines are rejected at compile time. - A fixed number of worker queues created up front. Go runs one goroutine per shard; C++ runs one worker thread per shard. - Go hashes the full routing key - account, account-group, or engine-wide - by independently mixing its numeric ID and kind. C++ hashes the account ID with a Fibonacci mix. - Routing is one multiply-shift per submit. There is no per-routing-key map lookup in Go or per-account map lookup in C++; synchronization is limited to ordering the submit against queue capacity and stop. Strengths: - The cheapest hot path: ~constant per submit. - O(1) memory regardless of how many distinct accounts are active. - Predictable worker count. Trade-offs: - A hot account saturates a single shard while the others stay idle. - No routing-queue created or removed observer signals (the shard queues exist for the engine lifetime). - In Go, distinct routing keys that hash to the same shard interleave through the same channel. In C++, different account IDs that hash to the same shard interleave through the same channel. Pick `Sharded` when the active account set is broad and roughly balanced or when raw throughput per submit matters more than routing-key isolation. ### Dynamic - Go: `asyncengine.NewBuilder(engine).Dynamic().MaxQueues(n).IdleCleanupAfter(d)` - C++: `TypedBuilder(driver).Dynamic().MaxQueues(n).IdleCleanupAfter(d)` - Go creates a queue on first submit for each account, account-group, or engine-wide routing key. C++ creates a per-account queue. - A background cleanup worker retires queues that have been empty and untouched for `IdleCleanupAfter`. - `MaxQueues` caps the live queue count. In Go, the count includes account, account-group, and engine-wide routing queues; submits for an unknown routing key past the cap return `ErrQueueLimit`. In C++, the count covers per-account queues and failures return `ErrorCode::QueueLimit`. `MaxQueues(0)` removes the cap. Strengths: - Full routing-key isolation in Go: a slow key does not starve others that happen to share a shard. C++ provides the same isolation per account. - Routing-queue observer signals (`OnQueueCreated`, `OnQueueRemoved`, latency); callback IDs are numeric projections and are not unique across routing kinds. - Memory scales with the active set rather than the total population. Trade-offs: - A synchronized map lookup on every submit. - A periodic cleanup worker. - Slightly higher per-queue memory because each queue holds its own bounded buffer and worker. Pick `Dynamic` when account activity is skewed, when routing-queue metrics with non-unique cross-kind numeric IDs are sufficient, or when the population is large enough that statically allocating shards would be wasteful. `MaxQueues` defaults to `runtime.NumCPU() * 32`, a value designed to be effectively non-restrictive on typical hosts while still bounding pathological growth (e.g. ephemeral per-request accounts). In Go, the limit counts all routing queues: `MaxQueues(n)` permits n usable account, account-group, or engine-wide queues in total. In C++, it counts account queues only. In both bindings, `MaxQueues(1)` permits one queue. Override with `MaxQueues(0)` for unbounded growth. ## Result Delivery: Future Every asynchronous operation returns a future (`pkg/future` in Go, `openpit::asyncengine::Future` in C++) that resolves exactly once. Ordinary validation and queue-submission failures may resolve synchronously. Not every failure before enqueue does: a call that hands off ownership of a handle for mandatory cleanup resolves only after that cleanup completes. This includes `AsyncRequest.Execute` and `AsyncRequest.Close`, plus the `AsyncReservation` and `AsyncDropCopyOperation` `Close`, `CommitAndClose`, and `RollbackAndClose` calls. A Go reentrancy refusal transfers no ownership: when the caller's context still carries an active lane marker for the target engine, the future resolves synchronously with `ErrReentrantLane` and no cleanup runs, leaving the handle with the caller. A context retained after its chain has finished no longer refuses. Go operations that mirror one return value use `future.Future[T]`; tuple-shaped operations use `future.Future2[A, B]`. C++ typed operations return `Future`, `Future`, or the concrete result type. `AsyncRequest::Execute` uses `PairFuture` for its reservation-or-rejects pair. Drop copy carries the same accepted-or-rejected pair as the main stage: `future.Future2[*AsyncDropCopyOperation, []reject.Reject]` in Go and `Future>` in C++, whose outcome holds a `std::shared_ptr>` and the rejects. The accepted reservation and drop-copy values are async wrappers, not direct-engine handles. In both bindings, `AsyncReservation` and `AsyncDropCopyOperation` wrap their direct-engine counterparts. Their terminal paths are `CommitAndClose`, `RollbackAndClose`, or `Close`, and finalization re-enters the producing account queue. Both wrappers synchronously forward `Lock` and `AccountAdjustments`. The drop-copy wrapper also forwards `AccountBlock` and `IsAccountBlocked`. These reads do not enqueue a separate engine call. Snapshot reads have three lifecycle states: - **Open:** a read proceeds. A read admitted while open completes before native finalization starts. - **Finalizing:** a new read fails immediately with `asyncengine.ErrFinalizationInProgress` in Go or `openpit::FinalizationInProgressError` in C++. This signal is transient: it does not transfer ownership, and the caller still owes a terminal call. For a closing finalization, a later read does not become successful when finalization finishes; it receives the terminal closed error instead. - **Closed:** Go returns `pretrade.ErrReservationClosed` or `pretrade.ErrDropCopyOperationClosed`. C++ throws a plain `openpit::Error` with the wrapper-specific closed message. This state is permanent: a later non-closing finalization cannot reopen the wrapper. Go future operations: - `f.Await(ctx)` - block until resolved or ctx fires. `Future[T]` yields `(T, error)`; `Future2[A, B]` yields `(A, B, error)`. - `f.Done()` - non-blocking check. - `f.TryGet()` - non-blocking read. - `f.Wait()` - channel that closes on resolution (for `select`). The future and the wrapped engine objects are decoupled: cancelling the context passed to `Await` does not stop the underlying engine call; it only stops the caller from waiting. In Go, call `Await` again or use `TryGet` after resolution and finish any caller-owned handle that eventually arrives: - `StartPreTrade` may yield an `AsyncRequest`; call `Execute` or `Close`. If `Execute` accepts the order, finish the resulting `AsyncReservation` too. - `ExecutePreTrade` or `AsyncRequest.Execute` may yield an `AsyncReservation`; call `CommitAndClose`, `RollbackAndClose`, or `Close`. - `ApplyDropCopy` may yield an `AsyncDropCopyOperation`; call `CommitAndClose`, `RollbackAndClose`, or `Close`. Do not resubmit an engine operation merely because its `Await` was cancelled. Abandoning the future is not a commit or rollback. In C++, `Await()` returns the value or throws the carried async error on the waiting thread; `Await(timeout)` returns `std::nullopt` when the timeout expires. For a move-only value, one `Await` call consumes it. A later `Await()` or `Await(timeout)` throws `ErrorCode::ValueConsumed`. The underlying task continues in both cases. ### Submission Timeout and Post-Trade Recovery The submission bound controls only the wait for queue space, not the engine operation once the task is enqueued. In Go, a submission `context.Context` that is already canceled is rejected before enqueue. While waiting for queue space, the slot and `ctx.Done()` can become ready together; Go may select the enqueue. Only the stored future completion is authoritative. An error returned because `Future.Await`'s own wait context ended says nothing about enqueue; inspect `TryGet` or await again with a separate live context. With the standard Engine driver, a stored completion of `context.Canceled` or `context.DeadlineExceeded` comes from the submission path and means enqueue did not happen, while a normal stored result means it did even if the submission context is canceled by then. A capped `Dynamic` strategy can store `ErrQueueLimit`. Submitting to a stopped engine, or a hard stop aborting a task before its engine call starts, stores `ErrStopped`. Custom driver errors also propagate unchanged. To preserve this retry provenance, an executed custom-driver call must not return any of those four submission-control errors, including wrappers matched by `errors.Is`; it must use a distinct execution error instead. In C++, a positive submission `timeout` can fail before enqueue with `ErrorCode::SubmitCancelled`. A capped `Dynamic` strategy can fail with `ErrorCode::QueueLimit`. Submitting to a stopped engine, or a hard stop aborting a task before its engine call starts, fails with `ErrorCode::Stopped`. For `ApplyDropCopy`, `ApplyExecutionReport`, and `ApplyAccountAdjustment`, any of those submission or stop errors before the engine call means that an already-occurred fact was not registered with the engine. The caller MUST retry. For an execution report, failure to retry leaves its reservation unreleased and its spot funds unsettled. The C++ value overloads of `ApplyDropCopy` and `ApplyExecutionReport` copy an lvalue or const rvalue and move a non-const rvalue. Pass an lvalue when a retry may be required so the caller retains the original. A named const object cast to an rvalue is copied, but only that named original remains available for retry. An unnamed const rvalue temporary is destroyed at the end of the full expression and does not provide a retained payload. Copying an execution report deep-clones a fill's pre-trade lock through the C ABI when one is present and can therefore throw. For those value overloads, a payload whose decayed static type is exactly the root `openpit::Order` or `openpit::ExecutionReport` is rejected at compile time. A reference whose static type is a concrete intermediate base such as `openpit::model::Order` or `openpit::model::ExecutionReport`, but whose dynamic type is more derived, throws `openpit::Error` synchronously before ownership transfer. A type-erased caller must use the corresponding `std::unique_ptr` or `std::unique_ptr` overload. Those overloads always transfer ownership. Before submission, retain the upstream payload or an independent copy of the exact concrete value; otherwise a submission or stop failure consumes the only payload and leaves nothing to retry. For C++ `ApplyAccountAdjustment`, pass an lvalue batch to its by-value parameter to retain the original for retry. The parameter is moved into the queued task, so a pre-enqueue failure destroys its copy; passing `std::move(...)` forfeits the caller's batch. The default non-positive timeout waits indefinitely for queue space, but a hard stop can still abort an already-enqueued task before it starts. Retention is therefore required even with the default timeout. ## Threading `AsyncEngine` requires the wrapped engine to be built with `AccountSync`. Per-account correctness holds in both strategies: no two operations for the same account ever run inside the engine concurrently, and within one account every queued operation runs in submit order. Go account operations use account routing keys. Account-group membership uses the first supplied account's key; group blocking and group currency operations use account-group routing keys; `UnblockAll` uses the engine-wide key. Go `AsyncEngine.RetireAccount` runs in the account's lane. The caller must still coordinate work submitted through other lanes or directly to the engine; see [Account Retirement](Account-Retirement.md). Parallelism across different routing keys depends on the strategy. `Dynamic` gives full routing-key isolation, so distinct keys are always processed in parallel. `Sharded` gives per-shard serialization: distinct keys that hash to the same shard share one worker and are processed one after another (a hot key can block others on its shard). The C++ facade has account routing only. Neither relaxes the per-account invariant above. The [Threading Contract](Threading-Contract.md) of the engine itself is respected end-to-end. `AsyncRequest`, `AsyncReservation`, and `AsyncDropCopyOperation` keep the same per-account queue across the boundary. Executing a started request, or committing, rolling back, or closing a reservation or drop-copy operation through its wrapper, re-enters the same per-account chain. That routing covers explicit calls only. Neither wrapper installs a destructor hook: Go has no finalizer at all, and the C++ wrapper's implicit release runs when the last `shared_ptr` owner goes away, on whatever thread drops it. So releasing a reservation or drop-copy operation without an explicit `Close`-flavored call performs its implicit rollback off the account queue. Finish every wrapper with `CommitAndClose`, `RollbackAndClose`, or `Close` when that lane matters. Finalizing through a wrapper does not change what a mutation finalizer owes the engine. In Go, mutation finalizers are void: a failure or panic is reported to core, arms the engine kill switch, and does not become an async finalizer future error. In C++, a callback exception both arms the kill switch and makes the async wrapper future fail with `ErrorCode::TaskFailed`. Because every mutation a Go or C++ policy registers is a custom-policy mutation, the block covers every account. Drop copy is not stopped by the block, and work already in flight on another lane may have passed the entry check before the switch was armed. Go can clear the block in the lane with `asyncEngine.Accounts().UnblockAll(ctx)`; in C++ the typed async engine uses `AsyncAccounts::UnblockAll`. See [Account Blocking](Account-Blocking.md#mutation-finalizer-contract). `ApplyDropCopy` uses the same account queue as every other operation for that account. A readable account ID is required: an order that does not expose one is refused up front without being queued. Go resolves the future immediately with `ErrMissingAccountID`. C++ resolves the future immediately with `ErrorCode::MissingAccountId`. `StartPreTrade` and `ExecutePreTrade` already behave this way. The core would reject such an order with `MissingRequiredField` before any policy callback runs, so there is nothing to gain by queueing it. Existing account and account-group blocks do not prevent a queued drop copy from reaching the engine. Go also refuses an uninitialized `param.AccountGroupID` before queuing a direct account-group operation or a chain sourced by that group. Those paths return `param.ErrUninitializedAccountGroupID`, the same error the synchronous account-group currency and market-data calls return; the synchronous registration and blocking calls leave an unset group to the engine, which rejects it as the reserved default group. Handle `0` is a legal, initialized group that selects the global default currency tier, so an unset group must be rejected at the public boundary rather than caught lower in the engine. C++ has no equivalent error: its `AccountGroupId` has no default constructor, so an unset group does not compile. The C++ typed facade is non-copyable but move-constructible. Its dispatch state keeps a stable address, so already-issued request and reservation wrappers stay valid after a move. Move assignment is disabled because replacing a live target could invalidate wrappers created by that target. Keep the resulting engine alive until all wrappers are released. ## Stop Modes - Go `StopGraceful(ctx)` refuses new submissions and waits for every already-queued task to run to completion. Returns `ctx.Err()` if ctx fires before workers drain; the engine is then partially stopped and `StopHard` may be invoked to complete the shutdown. - Go `StopHard(ctx)` refuses new submissions, aborts every task that has not yet started with `ErrStopped`, and waits for the currently-running task in each worker to finish. Returns `ctx.Err()` if ctx fires before the in-flight task in each worker finishes. - C++ `StopGraceful(timeout)` and `StopHard(timeout)` have the same queue behavior and return `true` when workers drained before the deadline. A hard stop resolves queued tasks with `ErrorCode::Stopped`. When the builder was wired via `AccountSyncReadyEngineBuilder.BuildAsync`, the underlying `*Engine` is released when the stop completes successfully (returns nil). If `StopGraceful` or `StopHard` returns `ctx.Err()` (timeout), the engine is not yet released; the caller must complete the shutdown (call `StopHard`) before the release happens. When the caller used `asyncengine.NewBuilder(engine)` and did not wire `WithStopUnderlying`, the engine handle remains owned by the caller. ## Observer `asyncengine.Observer` in Go and `openpit::asyncengine::Observer` in C++ are optional diagnostic interfaces. The default is `NoopObserver`. Wire only the methods you need - the package itself has zero external observability dependencies (no OpenTelemetry, Prometheus, or logging in the import graph), so you decide where the signals go. Available callbacks: `OnEnqueue`, `OnDequeue`, `OnComplete`, `OnSlowSubmit`, `OnQueueFullBlocked`, `OnQueueCreated`, `OnQueueRemoved`, `OnSubmitCancelled`. Observer callbacks are diagnostic only. In C++, exceptions thrown by an observer are ignored so observability cannot change queue or task semantics. Callback ID semantics differ by binding. In Go, each callback projects the routing key's numeric ID into `AccountID`: account lanes carry the account ID, account-group lanes carry the group ID, and the engine-wide lane carries 0. Equal numeric IDs are not unique across routing kinds, so `AccountID(0)` is not account-zero-only in Go. In C++, every callback ID is a real account ID, and `AccountID(0)` means account zero only. ## Example ```go package main import ( "context" "log" "time" "go.openpit.dev/openpit" "go.openpit.dev/openpit/model" "go.openpit.dev/openpit/param" "go.openpit.dev/openpit/pretrade/policies" ) func main() { // Build an AccountSync engine and wrap it into an async facade in one // chain. BuildAsync is only available on the AccountSync builder; for // FullSync or NoSync engines the bundled async helper is not used. asyncBuilder, err := openpit.NewEngineBuilder(). AccountSync(). Builtin(policies.BuildOrderValidation()). Builtin( policies.BuildRateLimit().BrokerBarrier( policies.RateLimitBrokerBarrier{ Limit: policies.RateLimit{ MaxOrders: 100, Window: time.Second, }, }, ), ). BuildAsync() if err != nil { log.Fatal(err) } // Pick a dispatch strategy. Use Sharded for cheap routing across a // balanced routing-key population; use Dynamic for routing-key isolation // and routing-queue metrics. async, err := asyncBuilder.Dynamic(). MaxQueues(0). IdleCleanupAfter(5 * time.Minute). Build() if err != nil { log.Fatal(err) } defer func() { if err := async.StopGraceful(context.Background()); err != nil { log.Printf("StopGraceful: %v", err) } }() // Submit a start-stage call. The future resolves once the worker has // executed the call. AsyncRequest.Execute and Close are queued in the // same per-account chain so AccountSync is never violated. usd, err := param.NewAsset("USD") if err != nil { log.Fatal(err) } aapl, err := param.NewAsset("AAPL") if err != nil { log.Fatal(err) } order := model.NewOrder() op := order.EnsureOperationView() op.SetInstrument(param.NewInstrument(aapl, usd)) op.SetAccountID(param.NewAccountIDFromUint64(99224416)) op.SetSide(param.SideBuy) price, err := param.NewPriceFromString("185") if err != nil { log.Fatal(err) } qty, err := param.NewQuantityFromString("100") if err != nil { log.Fatal(err) } op.SetTradeAmount(param.NewQuantityTradeAmount(qty)) op.SetPrice(price) request, rejects, err := async.StartPreTrade( context.Background(), order, ).Await(context.Background()) if err != nil { log.Fatal(err) } if request == nil { // Rejected at the start stage; inspect rejects. _ = rejects return } // The async request preserves AccountSync across the Start - Execute // boundary by routing Execute through the same per-account queue. The // future yields the same (reservation, rejects, error) tuple the // synchronous main stage returns. reservation, rejects, err := request.Execute( context.Background(), ).Await(context.Background()) if err != nil { log.Fatal(err) } if reservation == nil { _ = rejects return } if _, err := reservation.CommitAndClose( context.Background(), ).Await(context.Background()); err != nil { log.Fatal(err) } } ``` ## Submitting Caller-Owned Work In Go, `Submit(ctx, accountID, fn)` enqueues caller-owned work into the same per-account queue. In C++, `Submit(accountId, std::function, timeout)` enqueues caller-owned work into the same per-account queue. Use it to run client-side work atomically with respect to engine calls on the same account - for example, to persist an order before `Execute`, or update a strategy book after `Commit`. Splitting a logical transaction into two `Submit` calls "surfaces" between them: tasks for the same account from elsewhere in the system can interleave between the two halves. To keep that boundary closed, bundle the work into a single `Submit` whose `fn` does both halves. ## Chains Chains are a Go-binding feature that replaces the manual `Submit` pattern when one account-lane task must combine caller work with several engine operations. The C++ binding has no equivalent. `Chain(source, begin)` makes one `ChainBuilder[State]`: a `model.Order` source routes by that order's account and enables order-only steps, while a `param.AccountID` source enables report and account-adjustment steps. `begin` creates the one caller-owned `State` inside the account lane, with its type inferred from the function. `ChainBuilder.RetireAccount` retires the source account in that lane and runs its `OnRetired` hook; a hook error fails the chain after retirement succeeds. Use named hook structs for operations that produce engine outcomes: `PreTradeHooks` (`OnRejected`, `OnReserved`), `DropCopyHooks` (`OnRejected`, `OnApplied`), `ExecutionReportHooks` (`Report`, `OnSettled`), and `AccountAdjustmentHooks` (`Adjustments`, `OnAdjusted`). `CheckOrder` and `Then` take functions directly. An accepted-operation hook returns `DecisionCommit` to apply the operation or `DecisionRollback` to end the chain. An unsupported decision also rolls the operation back and fails the chain. Hooks receive caller `State` plus only a narrow operation result, never an engine, driver, queue, future, or lifecycle object. `OperationResult`, `BlockingOperationResult`, and `DropCopyResult` are readable only while their hook runs; after that their reads return `ErrChainResultUnavailable`. `ApplyDropCopy` does not stop because an account was already blocked. `DropCopyResult.IsAccountBlocked` returns the apply-time blocked-state snapshot for the order account, including a block that existed before drop copy ran; the snapshot is captured before `ApplyDropCopy` returns and does not track later registry changes. `OrderCheckResult` keeps `Passed` and `Rejects` readable after its hook, but its operation reads have the same hook-only lifetime. A hook context retains the submitted context's values but not its cancellation or deadline. The submitted context bounds queue admission only. `Chain.CheckOrder` is the only asynchronous full-pipeline dry-run path. Start-stage dry run remains synchronous-only. A failed verdict itself does not end the chain. `Run` resolves a `ChainOutcome` with or without a terminal hook, and its `Err` agrees with the error returned by `Await` whenever `Await` returned an outcome. When `Await` returns `ctx.Err()` instead, it yields the zero `ChainOutcome` - status `ChainOutcomeUnknown`, no `Err`, `RetryUnsafe` false - and that outcome must not be interpreted. It reports only that the wait ended: waiting never cancels the chain, and the chain may still be running or may already have finished, so the caller re-awaits or uses `TryGet`. The outcome passed to `Finally` describes the chain before that hook runs; the outcome resolved into the future describes the whole run, including `Finally`. `Finally` returns a `ChainRunner`, and that runner's `Run` is the only way to run a chain with a terminal hook. `Finally` runs after a successful `begin` and a started chain - on completion, rejection, failure, or panic - after native cleanup and before the future resolves. If `begin` fails or panics, there is no State and `Finally` does not run. Validation and queue-submission failures, plus any pre-start abort including hard stop and idle-queue retirement or drain, invoke no chain callbacks: `Begin`, step hooks, and `Finally` do not run. `AsyncEngine` observer callbacks may still run. Mandatory cleanup cannot live in `Finally` alone. `Run` itself carries that guarantee: with or without a terminal hook, it resolves the `ChainOutcome` only after every engine call, hook, and required handle cleanup - a pending reservation or drop-copy operation is already rolled back and closed by the time the future resolves. A whole chain is one account-lane task, so its caller work and engine calls cannot interleave with other work for that account. It is not atomic across steps, so the caller tracks its own progress in `State`. `RetryUnsafe` is conservative: it reports that the chain entered an engine call that can change state, not that a mutation is known to have happened. It is meaningful on `ChainOutcomeCompleted`, `ChainOutcomeRejected`, and `ChainOutcomeFailed`; on a rejected or completed outcome it is the only signal that rerunning may be unsafe, because neither carries an error to wrap the sentinel in. It is not meaningful on `ChainOutcomeUnknown`, the zero value `Await` yields on `ctx.Err()`: its `RetryUnsafe: false` is never permission to rerun a chain that may still be running or may already have changed state. On a failed outcome, `Err` wraps `ErrChainRetryUnsafe` when the SDK cannot prove nothing changed. Rerunning is unsafe; the sentinel does not claim a mutation happened. In Go, submitting another operation to the same engine with a context that contains an active lane marker for that engine anywhere in its marker stack, including below inactive or other-engine inner markers, resolves its future with `ErrReentrantLane` rather than waiting on the hook's own lane. A retained context whose target-engine marker has become inactive no longer refuses. The marker is diagnostic only: caller code that supplies an unrelated context bypasses it. A hook must never synchronously re-enter the engine with any context, because its running chain occupies the account lane. ```go package main import ( "context" "fmt" "log" "go.openpit.dev/openpit" "go.openpit.dev/openpit/asyncengine" "go.openpit.dev/openpit/model" "go.openpit.dev/openpit/param" "go.openpit.dev/openpit/pretrade" "go.openpit.dev/openpit/pretrade/policies" "go.openpit.dev/openpit/reject" ) func main() { asyncBuilder, err := openpit.NewEngineBuilder(). AccountSync(). Builtin(policies.BuildOrderValidation()). BuildAsync() if err != nil { log.Fatal(err) } async, err := asyncBuilder.Dynamic().Build() if err != nil { log.Fatal(err) } defer func() { if err := async.StopGraceful(context.Background()); err != nil { log.Printf("StopGraceful: %v", err) } }() aapl, err := param.NewAsset("AAPL") if err != nil { log.Fatal(err) } usd, err := param.NewAsset("USD") if err != nil { log.Fatal(err) } accountID := param.NewAccountIDFromUint64(99224416) instrument := param.NewInstrument(aapl, usd) order := model.NewOrder() orderOperation := order.EnsureOperationView() orderOperation.SetAccountID(accountID) orderOperation.SetInstrument(instrument) orderOperation.SetSide(param.SideBuy) quantity, err := param.NewQuantityFromString("100") if err != nil { log.Fatal(err) } orderOperation.SetTradeAmount(param.NewQuantityTradeAmount(quantity)) price, err := param.NewPriceFromString("185") if err != nil { log.Fatal(err) } orderOperation.SetPrice(price) type state struct { report model.ExecutionReport reservationLock pretrade.Lock instrument param.Instrument accountID param.AccountID adjustmentCount int settled bool persisted bool } runner := asyncengine.Chain(order, func(context.Context) (*state, error) { return &state{ accountID: accountID, instrument: instrument, }, nil }). ExecutePreTrade(asyncengine.PreTradeHooks[*state]{ OnRejected: func( context.Context, *state, []reject.Reject, ) error { return fmt.Errorf("pre-trade order was rejected") }, OnReserved: func( _ context.Context, got *state, result asyncengine.OperationResult, ) (asyncengine.Decision, error) { lock, err := result.Lock() if err != nil { return asyncengine.DecisionRollback, err } adjustments, err := result.AccountAdjustments() if err != nil { return asyncengine.DecisionRollback, err } got.reservationLock = lock got.adjustmentCount = len(adjustments) return asyncengine.DecisionCommit, nil }, }). ApplyExecutionReport(asyncengine.ExecutionReportHooks[*state]{ Report: func( _ context.Context, got *state, ) (model.ExecutionReport, error) { report := model.NewExecutionReport() reportOperation := report.EnsureOperationView() reportOperation.SetAccountID(got.accountID) reportOperation.SetInstrument(got.instrument) reportOperation.SetSide(param.SideBuy) got.report = report return report, nil }, OnSettled: func( _ context.Context, got *state, _ pretrade.PostTradeResult, ) error { got.settled = true return nil }, }). Then(func(_ context.Context, got *state) error { // Persist caller-owned metadata after the execution report settles. got.persisted = true return nil }). Finally(func( _ context.Context, got *state, outcome asyncengine.ChainOutcome, ) error { if outcome.Status != asyncengine.ChainOutcomeCompleted { return fmt.Errorf("chain finished with %v", outcome.Status) } log.Printf("settled=%t persisted=%t", got.settled, got.persisted) return nil }) chainFuture := runner.Run(context.Background(), async) // context.Background cannot be cancelled, so Await always yields a verdict // and ChainOutcomeUnknown is unreachable here. With a bounded context, // Unknown means only that the wait ended: the chain may still be running or // may already have finished. Re-await with another caller-controlled context // or use TryGet before interpreting the outcome. outcome, err := chainFuture.Await(context.Background()) if outcome.Status == asyncengine.ChainOutcomeUnknown { log.Fatal("chain Await() returned no outcome with a non-cancellable context") } if err != nil { if outcome.RetryUnsafe { log.Fatalf("chain failed and must not be retried: %v", err) } else { log.Fatalf("chain failed before retry became unsafe: %v", err) } } log.Printf( "chain status=%v retryUnsafe=%t", outcome.Status, outcome.RetryUnsafe, ) } ``` ## Lifecycle Notes - Aborted tasks resolve their future with `ErrStopped` in Go and `ErrorCode::Stopped` in C++. - In both bindings, `AsyncReservation` and `AsyncDropCopyOperation` methods that "and close" (`Close`, `CommitAndClose`, `RollbackAndClose`) release the underlying handle as a safety net even on abort. An aborted plain `Commit` or `Rollback` does not release it, so the caller must still close the wrapper. - In both bindings, an aborted `AsyncRequest.Execute` releases the underlying request before resolving the future. The caller does not need a separate cleanup path. - The Go `ctx` and C++ `timeout` passed to a submit method control how long the producer waits for queue space. Once queued, the worker runs or aborts the task. ## Related Pages - [Threading Contract](Threading-Contract.md): per-mode sync contract that AsyncEngine respects. - [Pre-trade Pipeline](Pre-trade-Pipeline.md): the Request/Reservation lifecycle that AsyncRequest/AsyncReservation wraps. - [Getting Started](Getting-Started.md): the synchronous flow that AsyncEngine layers on top of. - [Account Blocking](Account-Blocking.md): the engine-wide block and the mutation finalizer contract that queued finalization is subject to.