Skip to content

Account Adjustments

Jump to bottom
Eugene Palchukovsky edited this page Jul 11, 2026 · 13 revisions

Account Adjustments

Account adjustment models non-trade operations (NTO): direct state corrections that do not originate from order execution.

Typical examples:

  • initial account load;
  • balance top-up or withdrawal;
  • position correction after a corporate action;
  • settlement or funding transfer represented as an explicit adjustment batch.

Balance vs Position Operations

  • Account adjustment balance operation: physical asset balance correction for one asset.
  • Account adjustment position operation: derivatives-like position correction for one instrument + collateral asset.

Both forms may include optional amount and bounds groups:

  • Account adjustment amount: balance, held, incoming.
  • Account adjustment bounds: inclusive lower/upper constraints for each amount field.

Batch Semantics

Apply account adjustments validates the input as an atomic batch:

  • evaluation order is deterministic: per-adjustment slice order, then per-policy registration order;
  • validation stops at the first reject;
  • outcome is all-or-nothing - a reject means the full batch is rejected;
  • on reject, the caller receives the failed index together with the policy reject payload.

Operational guidance:

  • If external logic needs to correct policy-related state while the engine is active, prefer apply account adjustments over direct parallel mutation of custom-policy fields.
  • Direct parallel access to custom-policy state still requires host-side synchronization.

Outcomes

A successful batch returns one outcome per asset each adjustment touched. Every outcome carries a signed delta and an absolute snapshot. Use the delta to keep your own ledger aligned with the engine; treat absolute as diagnostic only. See Balance Reconciliation for the full delta-versus-absolute contract and a scheme for persisting limits in your own store.

Examples

Go 
accountID := param.NewAccountIDFromUint64(99224416)

usd, err := param.NewAsset("USD")
if err != nil {
  panic(err)
}
spx, err := param.NewAsset("SPX")
if err != nil {
  panic(err)
}
entryPrice, _ := param.NewPriceFromString("95000")
totalCash, _ := param.NewPositionSizeFromString("10000")
totalPosition, _ := param.NewPositionSizeFromString("-3")

cashAdj, err := model.NewAccountAdjustmentFromValues(model.AccountAdjustmentValues{
  BalanceOperation: optional.Some(
    model.NewAccountAdjustmentBalanceOperationFromValues(
      model.AccountAdjustmentBalanceOperationValues{
        Asset: optional.Some(usd),
      },
    ),
  ),
  Amount: optional.Some(
    model.NewAccountAdjustmentAmountFromValues(model.AccountAdjustmentAmountValues{
      Balance: optional.Some(param.NewAbsoluteAdjustmentAmount(totalCash)),
    }),
  ),
})
if err != nil {
  panic(err)
}

posAdj, err := model.NewAccountAdjustmentFromValues(model.AccountAdjustmentValues{
  PositionOperation: optional.Some(
    model.NewAccountAdjustmentPositionOperationFromValues(
      model.AccountAdjustmentPositionOperationValues{
        Instrument:        optional.Some(param.NewInstrument(spx, usd)),
        CollateralAsset:   optional.Some(usd),
        AverageEntryPrice: optional.Some(entryPrice),
        Mode:              optional.Some(param.PositionModeHedged),
      },
    ),
  ),
  Amount: optional.Some(
    model.NewAccountAdjustmentAmountFromValues(model.AccountAdjustmentAmountValues{
      Balance: optional.Some(param.NewAbsoluteAdjustmentAmount(totalPosition)),
    }),
  ),
})
if err != nil {
  panic(err)
}

// On accept, rejects.IsSet() is false; the second result holds the
// per-asset account-adjustment outcomes.
rejects, _, err := engine.ApplyAccountAdjustment(
  accountID,
  []model.AccountAdjustment{cashAdj, posAdj},
)
_ = rejects
_ = err
Python 
import openpit
import openpit.pretrade.policies

# Build one batch that mixes balance and position adjustments.
account_id = openpit.param.AccountId.from_int(99224416)

adjustments = [
    openpit.AccountAdjustment(
        operation=openpit.AccountAdjustmentBalanceOperation(asset="USD"),
        amount=openpit.AccountAdjustmentAmount(
            balance=openpit.param.AdjustmentAmount.absolute(
                openpit.param.PositionSize(10000)
            )
        ),
    ),
    openpit.AccountAdjustment(
        operation=openpit.AccountAdjustmentPositionOperation(
            instrument=openpit.Instrument("SPX", "USD"),
            collateral_asset="USD",
            average_entry_price=openpit.param.Price(95000),
            mode=openpit.param.PositionMode.HEDGED,
        ),
        amount=openpit.AccountAdjustmentAmount(
            balance=openpit.param.AdjustmentAmount.absolute(
                openpit.param.PositionSize(-3)
            )
        ),
    ),
]

# The engine validates the whole batch atomically.
engine = (
    openpit.Engine.builder()
    .no_sync()
    .builtin(openpit.pretrade.policies.build_order_validation())
    .build()
)
result = engine.apply_account_adjustment(
    account_id=account_id,
    adjustments=adjustments,
)
assert result.ok
JavaScript 
import { Engine } from "@openpit/engine";
import { type AccountAdjustmentInit } from "@openpit/engine/model";
import { AdjustmentAmount } from "@openpit/engine/param";
import { buildOrderValidation } from "@openpit/engine/pretrade/policies";

// Build one batch that mixes balance and position adjustments. Each adjustment
// is a plain object literal; position sizes cross the boundary as decimal
// strings.
const accountId = 99224416;

const adjustments: AccountAdjustmentInit[] = [
  {
    operation: { asset: "USD" },
    amount: { balance: AdjustmentAmount.absolute("10000") },
  },
  {
    operation: {
      underlyingAsset: "SPX",
      settlementAsset: "USD",
      collateralAsset: "USD",
      averageEntryPrice: "95000",
      mode: "hedged",
    },
    amount: { balance: AdjustmentAmount.absolute("-3") },
  },
];

// The engine validates the whole batch atomically.
const engine = Engine.builder()
  .builtin(buildOrderValidation())
  .build();
const result = engine.applyAccountAdjustment(accountId, adjustments);
C++ 
#include <cassert>

namespace aa = openpit::accountadjustment;
namespace param = openpit::param;
namespace policies = openpit::pretrade::policies;

// Build one batch that mixes balance and position adjustments.
const param::AccountId accountId = param::AccountId::FromUint64(99224416);

aa::AccountAdjustment cashAdj;
{
  aa::BalanceOperation balance;
  balance.asset = "USD";
  cashAdj.operation = aa::Operation::OfBalance(std::move(balance));
  aa::Amount amount;
  amount.balance =
      param::AdjustmentAmount::OfAbsolute(param::PositionSize::FromString("10000"));
  cashAdj.amount = std::move(amount);
}

aa::AccountAdjustment posAdj;
{
  aa::PositionOperation position;
  position.instrument = openpit::model::Instrument("SPX", "USD");
  position.collateralAsset = "USD";
  position.averageEntryPrice = param::Price::FromString("95000");
  position.mode = openpit::model::PositionMode::Hedged;
  posAdj.operation = aa::Operation::OfPosition(std::move(position));
  aa::Amount amount;
  amount.balance =
      param::AdjustmentAmount::OfAbsolute(param::PositionSize::FromString("-3"));
  posAdj.amount = std::move(amount);
}

const std::vector<aa::AccountAdjustment> adjustments{std::move(cashAdj),
                                                     std::move(posAdj)};

// The engine validates the whole batch atomically.
openpit::EngineBuilder builder(openpit::SyncPolicy::None);
builder.Add(policies::OrderValidationPolicy{});
const openpit::Engine engine = builder.Build();

// On accept the result passes and carries the per-asset account-adjustment
// outcomes.
const openpit::AdjustmentResult result =
    engine.ApplyAccountAdjustment(accountId, adjustments);
assert(result.Passed());
Rust 
use openpit::param::{
    AccountId, AdjustmentAmount, Asset, PositionMode, PositionSize, Price,
};
use openpit::{
    AccountAdjustmentAmount, AccountAdjustmentBalanceOperation,
    AccountAdjustmentPositionOperation, Engine, Instrument,
};

#[derive(Clone)]
enum AccountAdjustmentOperation {
    Balance(AccountAdjustmentBalanceOperation),
    Position(AccountAdjustmentPositionOperation),
}

#[derive(Clone)]
struct AccountAdjustment {
    operation: AccountAdjustmentOperation,
    amount: AccountAdjustmentAmount,
}

// Build one batch that mixes balance and position adjustments.
let account_id = AccountId::from_u64(99224416);

let adjustments = vec![
    AccountAdjustment {
        operation: AccountAdjustmentOperation::Balance(
            AccountAdjustmentBalanceOperation {
                asset: Asset::new("USD")?,
                average_entry_price: None,
                realized_pnl: None,
            },
        ),
        amount: AccountAdjustmentAmount {
            balance: Some(AdjustmentAmount::Absolute(
                PositionSize::from_f64(10000.0)?,
            )),
            held: None,
            incoming: None,
        },
    },
    AccountAdjustment {
        operation: AccountAdjustmentOperation::Position(
            AccountAdjustmentPositionOperation {
                instrument: Instrument::new(
                    Asset::new("SPX")?,
                    Asset::new("USD")?,
                ),
                collateral_asset: Asset::new("USD")?,
                average_entry_price: Price::from_f64(95000.0)?,
                mode: PositionMode::Hedged,
                leverage: None,
            },
        ),
        amount: AccountAdjustmentAmount {
            balance: Some(AdjustmentAmount::Absolute(
                PositionSize::from_f64(-3.0)?,
            )),
            held: None,
            incoming: None,
        },
    },
];

struct AcceptAllAdjustments;

impl<Sync> openpit::pretrade::PreTradePolicy<(), (), AccountAdjustment, Sync>
    for AcceptAllAdjustments
where
    Sync: openpit::SyncMode,
{
    fn name(&self) -> &'static str {
        "AcceptAllAdjustments"
    }

    fn apply_account_adjustment(
        &self,
        _ctx: &openpit::AccountAdjustmentContext<
            <Sync as openpit::SyncMode>::StorageLockingPolicyFactory,
        >,
        _account_id: openpit::param::AccountId,
        _adjustment: &AccountAdjustment,
        _mutations: &mut openpit::Mutations,
    ) -> Result<Vec<openpit::AccountOutcomeEntry>, openpit::pretrade::Rejects> {
        Ok(Vec::new())
    }
}

// The engine validates the whole batch atomically.
let engine = Engine::builder::<(), (), AccountAdjustment>()
    .no_sync()
    .pre_trade(AcceptAllAdjustments)
    .build()?;
let result = engine.apply_account_adjustment(account_id, &adjustments);
assert!(result.is_ok());

Writing an Account Adjustment Policy with Rollback

Account adjustment policies can register rollback actions that undo intermediate state if a later element in the batch is rejected. The engine applies rollback actions in reverse registration order (last registered = first rolled back).

Rollback Safety

Account adjustment batches run within a single engine call. No external system (venue, risk aggregator) observes intermediate state between elements, so rollback by absolute value is safe: a policy can capture the current value before modification and restore it on rollback without risking inconsistency.

The pre-trade pipeline is different: a reservation may be observed by external systems between creation and finalization, so pre-trade policies should prefer delta-based rollback.

Example: Balance Limit Policy

The policy below tracks cumulative adjustment totals per asset and rejects the batch if any asset exceeds a configured limit. On rejection, all previously accumulated totals are rolled back to their state before the batch started.

Go 
type CumulativeLimitPolicy struct {
  maxCumulative param.Volume
  totals        map[string]param.Volume
}

func (v *CumulativeLimitPolicy) Close() {}

func (v *CumulativeLimitPolicy) Name() string { return "CumulativeLimitPolicy" }

func (v *CumulativeLimitPolicy) PolicyGroupID() model.PolicyGroupID {
    return model.DefaultPolicyGroupID
}

func (v *CumulativeLimitPolicy) CheckPreTradeStart(
  pretrade.Context,
  model.Order,
) []reject.Reject {
  return nil
}

func (v *CumulativeLimitPolicy) PerformPreTradeCheck(
  pretrade.Context,
  model.Order,
  tx.Mutations,
  pretrade.Result,
) []reject.Reject {
  return nil
}

func (v *CumulativeLimitPolicy) ApplyExecutionReport(
  pretrade.PostTradeContext,
  model.ExecutionReport,
  pretrade.PostTradeAdjustments,
) []reject.AccountBlock {
  return nil
}

func (v *CumulativeLimitPolicy) ApplyAccountAdjustment(
  _ accountadjustment.Context,
  accountID param.AccountID,
  adjustment model.AccountAdjustment,
  mutations tx.Mutations,
  outcomes pretrade.AccountOutcomes,
) []reject.Reject {
  _ = accountID
  _ = adjustment
  _ = outcomes
  return nil
}
Python 
import openpit


class CumulativeLimitPolicy(openpit.pretrade.Policy):
    """Tracks cumulative totals per asset, rejects batch on limit breach."""

    def __init__(self, max_cumulative: openpit.param.Volume) -> None:
        self._max = max_cumulative
        self._totals: dict[str, openpit.param.Volume] = {}

    @property
    def name(self) -> str:
        return "CumulativeLimitPolicy"

    def apply_account_adjustment(
        self,
        ctx: openpit.AccountAdjustmentContext,
        account_id: openpit.param.AccountId,
        adjustment: openpit.AccountAdjustment,
    ) -> (
        list[openpit.pretrade.PolicyReject]
        | tuple[openpit.Mutation, ...]
        | None
    ):
        del ctx, account_id
        # Use the asset as the aggregation key for the cumulative limit.
        asset_id = adjustment.operation.asset

        prev = self._totals.get(asset_id, openpit.param.Volume("0"))
        # Simplified - real code would add delta to prev.
        new_total = prev

        # Reject if limit breached.
        if new_total > self._max:
            return [
                openpit.pretrade.PolicyReject(
                    code=openpit.pretrade.RejectCode.RISK_LIMIT_EXCEEDED,
                    reason="cumulative limit exceeded",
                    details=f"{asset_id}: {new_total} > {self._max}",
                    scope=openpit.pretrade.RejectScope.ACCOUNT,
                )
            ]

        # Apply immediately so later adjustments in the same batch
        # see the updated total.
        self._totals[asset_id] = new_total

        # Rollback by absolute value - safe in account adjustment pipeline
        # because no external system sees intermediate batch state.
        prev_value = prev
        asset_key = asset_id
        return (
            openpit.Mutation(
                # Commit is empty: state was applied eagerly.
                commit=lambda: None,
                rollback=lambda: self._totals.__setitem__(asset_key, prev_value),
            ),
        )
JavaScript 
import { Engine } from "@openpit/engine";
import { type AccountAdjustmentContext } from "@openpit/engine/accountadjustment";
import { type AccountAdjustment } from "@openpit/engine/model";
import {
  AccountId,
  AdjustmentAmount,
  PositionSize,
} from "@openpit/engine/param";
import {
  type Policy,
  type PolicyAccountAdjustmentResult,
  type PolicyPreTradeResult,
  type PolicyReject,
} from "@openpit/engine/pretrade";

// Tracks cumulative totals per asset, rejects the batch on a limit breach.
class CumulativeLimitPolicy implements Policy {
  readonly name = "CumulativeLimitPolicy";
  private readonly totals = new Map<string, PositionSize>();

  constructor(private readonly maxCumulative: PositionSize) {}

  // The pre-trade hooks are required by the Policy interface; this policy only
  // acts on the account-adjustment path, so they accept with no contribution.
  checkPreTradeStart(): Iterable<PolicyReject> {
    return [];
  }

  performPreTradeCheck(): PolicyPreTradeResult {
    return {};
  }

  applyAccountAdjustment(
    _ctx: AccountAdjustmentContext,
    _accountId: AccountId,
    adjustment: AccountAdjustment,
  ): PolicyAccountAdjustmentResult {
    // Use the asset as the aggregation key for the cumulative limit.
    const operation = adjustment.operation;
    const assetId =
      operation !== undefined && "asset" in operation
        ? (operation.asset ?? "")
        : "";

    const previous = this.totals.get(assetId);
    const current = previous ?? PositionSize.fromInt(0n);
    const balance = adjustment.amount?.balance;
    if (balance === undefined) {
      return [];
    }

    const absolute = balance.asAbsolute;
    let newTotal: PositionSize;
    if (absolute !== undefined) {
      newTotal = absolute;
    } else {
      const delta = balance.asDelta;
      if (delta === undefined) {
        return [];
      }
      newTotal = current.add(delta);
    }

    // Reject if the limit is breached.
    if (newTotal.compare(this.maxCumulative) > 0) {
      return [
        {
          code: "RiskLimitExceeded",
          reason: "cumulative limit exceeded",
          details: `${assetId}: ${newTotal.toString()} > ${this.maxCumulative.toString()}`,
          scope: "account",
        },
      ];
    }

    // Apply immediately so later adjustments in the same batch see the updated
    // total.
    this.totals.set(assetId, newTotal);

    // Rollback by absolute value - safe in the account-adjustment pipeline
    // because no external system sees intermediate batch state. Commit is empty:
    // the state was applied eagerly.
    return [
      {
        commit: () => {},
        rollback: () => {
          if (previous === undefined) {
            this.totals.delete(assetId);
          } else {
            this.totals.set(assetId, previous);
          }
        },
      },
    ];
  }
}

// Seed an absolute value, then reject a batch after its first delta was
// applied. A final delta reaches the limit exactly only if the failed batch
// rolled its first mutation back.
const engine = Engine.builder()
  .preTrade(new CumulativeLimitPolicy(PositionSize.fromString("100")))
  .build();
const accountId = 99224416;
const adjustment = (
  amount: ReturnType<typeof AdjustmentAmount.absolute>,
) => ({
  operation: { asset: "USD" },
  amount: { balance: amount },
});

const seed = engine.applyAccountAdjustment(accountId, [
  adjustment(AdjustmentAmount.absolute("40")),
]);

const rejected = engine.applyAccountAdjustment(accountId, [
  adjustment(AdjustmentAmount.delta("30")),
  adjustment(AdjustmentAmount.delta("40")),
]);

const afterRollback = engine.applyAccountAdjustment(accountId, [
  adjustment(AdjustmentAmount.delta("60")),
]);

if (!seed.ok) {
  throw new Error("the absolute seed must be accepted");
}
if (
  rejected.ok ||
  rejected.failedIndex !== 1 ||
  rejected.rejects[0]?.code !== "RiskLimitExceeded"
) {
  throw new Error("the second delta must breach the cumulative limit");
}
if (!afterRollback.ok) {
  throw new Error("the failed batch must roll the first delta back");
}
C++

The C++ account-adjustment callback receives the current adjustment, a mutation collector, and an account-outcome collector. Apply policy-local state eagerly so later elements in the same batch see it, then register commit and rollback callbacks with tx::Mutations::Push. If any policy rejects an element, OpenPit runs the preceding rollbacks in reverse order and rejects the whole batch.

struct CumulativeLimitState {
  std::map<std::string, openpit::param::PositionSize> totals;
  std::size_t commits = 0;
  std::size_t rollbacks = 0;
};

// Tracks the last accepted absolute balance per asset. Each accepted element
// mutates eagerly so the next element in the same batch sees it, then registers
// commit/rollback callbacks with the engine transaction.
class CumulativeLimitPolicy {
 public:
  CumulativeLimitPolicy(
      openpit::param::PositionSize maxCumulative,
      std::shared_ptr<CumulativeLimitState> state)
      : m_maxCumulative(maxCumulative), m_state(std::move(state)) {}

  [[nodiscard]] std::string_view Name() const noexcept {
    return "CumulativeLimitPolicy";
  }

  [[nodiscard]] openpit::pretrade::PolicyDecision ApplyAccountAdjustment(
      const openpit::accountadjustment::Context& context,
      openpit::param::AccountId accountId,
      const openpit::accountadjustment::AccountAdjustment& adjustment,
      openpit::tx::Mutations& mutations,
      openpit::pretrade::AccountOutcomes& outcomes) const {
    static_cast<void>(context);
    static_cast<void>(accountId);
    static_cast<void>(outcomes);

    const auto* balance =
        adjustment.operation ? adjustment.operation->AsBalance() : nullptr;
    if (balance == nullptr || !balance->asset || !adjustment.amount ||
        !adjustment.amount->balance ||
        !adjustment.amount->balance->IsAbsolute()) {
      return {};
    }

    const std::string asset = *balance->asset;
    const openpit::param::PositionSize next =
        adjustment.amount->balance->Value();
    if (next > m_maxCumulative) {
      openpit::pretrade::PolicyDecision decision;
      decision.Push(openpit::pretrade::Reject(
          std::string(Name()), openpit::pretrade::RejectScope::Account,
          openpit::pretrade::RejectCode::RiskLimitExceeded,
          "cumulative limit exceeded",
          asset + " absolute balance exceeds the configured limit"));
      return decision;
    }

    std::optional<openpit::param::PositionSize> previous;
    if (const auto it = m_state->totals.find(asset);
        it != m_state->totals.end()) {
      previous = it->second;
    }
    m_state->totals.insert_or_assign(asset, next);

    const auto state = m_state;
    mutations.Push(
        [state] { ++state->commits; },
        [state, asset, previous] {
          ++state->rollbacks;
          if (previous) {
            state->totals.insert_or_assign(asset, *previous);
          } else {
            state->totals.erase(asset);
          }
        });
    return {};
  }

 private:
  openpit::param::PositionSize m_maxCumulative;
  std::shared_ptr<CumulativeLimitState> m_state;
};

[[nodiscard]] openpit::accountadjustment::AccountAdjustment AbsoluteBalance(
    std::string asset, std::string_view value) {
  openpit::accountadjustment::BalanceOperation balance;
  balance.asset = std::move(asset);

  openpit::accountadjustment::Amount amount;
  amount.balance = openpit::param::AdjustmentAmount::OfAbsolute(
      openpit::param::PositionSize::FromString(value));

  openpit::accountadjustment::AccountAdjustment adjustment;
  adjustment.operation =
      openpit::accountadjustment::Operation::OfBalance(std::move(balance));
  adjustment.amount = std::move(amount);
  return adjustment;
}

const auto state = std::make_shared<CumulativeLimitState>();
openpit::EngineBuilder builder(openpit::SyncPolicy::None);
openpit::pretrade::CustomPolicy<CumulativeLimitPolicy> policy(
    "CumulativeLimitPolicy",
    CumulativeLimitPolicy(openpit::param::PositionSize::FromString("1000000"),
                          state));
builder.Add(policy);
const openpit::Engine engine = builder.Build();

const openpit::param::AccountId accountId =
    openpit::param::AccountId::FromUint64(99224416);

const openpit::AdjustmentResult accepted = engine.ApplyAccountAdjustment(
    accountId, std::vector<openpit::accountadjustment::AccountAdjustment>{
                   AbsoluteBalance("USD", "100")});
assert(accepted.Passed());
assert(state->totals.at("USD").ToString() == "100");
assert(state->commits == 1);

const openpit::AdjustmentResult rejected = engine.ApplyAccountAdjustment(
    accountId, std::vector<openpit::accountadjustment::AccountAdjustment>{
                   AbsoluteBalance("USD", "200"),
                   AbsoluteBalance("USD", "2000000")});
assert(!rejected.Passed());
assert(rejected.batchError->FailedAdjustmentIndex() == 1);
assert(state->totals.at("USD").ToString() == "100");
assert(state->commits == 1);
assert(state->rollbacks == 1);
Rust 
use std::sync::Arc;

use openpit::param::{AccountId, Volume};
use openpit::pretrade::{Reject, RejectCode, RejectScope, Rejects};
use openpit::storage::{
    CreateStorageFor, LockingPolicyFactory, Storage, StorageBuilder,
};
use openpit::pretrade::PreTradePolicy;
use openpit::{AccountAdjustmentContext, Mutation, Mutations};

struct BalanceLimitPolicy<StorageLockingPolicyFactory>
where
    StorageLockingPolicyFactory: LockingPolicyFactory,
{
    max_total: Volume,
    totals: Arc<
        Storage<
            String,
            Volume,
            StorageLockingPolicyFactory::Policy,
        >,
    >,
}

impl<StorageLockingPolicyFactory>
    BalanceLimitPolicy<StorageLockingPolicyFactory>
where
    StorageLockingPolicyFactory: LockingPolicyFactory
        + CreateStorageFor<String>,
{
    fn new(
        max_total: Volume,
        storage_builder: &StorageBuilder<StorageLockingPolicyFactory>,
    ) -> Self {
        Self {
            max_total,
            totals: Arc::new(storage_builder.create_for_bound_key()),
        }
    }
}

/// Adjustment type must expose an asset and a delta amount.
trait HasAssetDelta {
    fn asset_id(&self) -> &str;
    fn delta(&self) -> Volume;
}

impl<Order, ExecutionReport, A, Sync, StorageLockingPolicyFactory>
    PreTradePolicy<Order, ExecutionReport, A, Sync>
    for BalanceLimitPolicy<StorageLockingPolicyFactory>
where
    A: HasAssetDelta,
    Sync: openpit::SyncMode,
    StorageLockingPolicyFactory: LockingPolicyFactory
        + CreateStorageFor<String>,
    StorageLockingPolicyFactory::Policy: 'static,
{
    fn name(&self) -> &str {
        "BalanceLimitPolicy"
    }

    fn apply_account_adjustment(
        &self,
        _ctx: &AccountAdjustmentContext<
            <Sync as openpit::SyncMode>::StorageLockingPolicyFactory,
        >,
        _account_id: AccountId,
        adjustment: &A,
        mutations: &mut Mutations,
    ) -> Result<Vec<openpit::AccountOutcomeEntry>, Rejects> {
        let asset_id = adjustment.asset_id().to_owned();
        let delta = adjustment.delta();

        let prev_total = self
            .totals
            .with(&asset_id, |total| *total)
            .unwrap_or(Volume::ZERO);

        let new_total = prev_total.checked_add(delta).map_err(|error| {
            Rejects::from(Reject::new(
                "BalanceLimitPolicy",
                RejectScope::Account,
                RejectCode::RiskLimitExceeded,
                "invalid adjustment total",
                error.to_string(),
            ))
        })?;

        if new_total > self.max_total {
            return Err(Rejects::from(Reject::new(
                "BalanceLimitPolicy",
                RejectScope::Account,
                RejectCode::RiskLimitExceeded,
                "cumulative adjustment exceeds limit",
                format!("asset {asset_id}: {new_total} > {}", self.max_total),
            )));
        }

        // Apply immediately so later adjustments in the same batch see the updated total.
        self.totals.with_mut(
            asset_id.clone(),
            || Volume::ZERO,
            |entry, _is_new| {
                *entry = new_total;
            },
        );

        // Register rollback: restore previous absolute value.
        // Safe because account adjustment batches are fully internal.
        let rollback_totals = Arc::clone(&self.totals);
        let rollback_asset = asset_id;

        mutations.push(Mutation::new(
            || {
                // Commit is empty: state was applied eagerly.
            },
            move || {
                // Rollback: restore absolute value captured before modification.
                rollback_totals.with_mut(
                    rollback_asset,
                    || Volume::ZERO,
                    |entry, _is_new| {
                        *entry = prev_total;
                    },
                );
            },
        ));

        Ok(Vec::new())
    }
}

Clone this wiki locally