Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 12 additions & 6 deletions src/application/handlers/block/candidateConfirmer.ts
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@ import { withTimeout } from "../../helpers/withTimeout";
import { bumpGeneratorsUpdatedAt } from "../../helpers/updatedAtBlock";
import { log } from "../../helpers/logger";
import { type DiscreteStatus } from "./shared";
import { refreshTwapExecutedTotals } from "../../helpers/executedAmounts";
import { refreshTwapExecutionState } from "../../helpers/executedAmounts";

// ─── CandidateConfirmer ──────────────────────────────────────────────────────
// Checks if candidate discrete orders exist on the Orderbook API.
Expand Down Expand Up @@ -142,10 +142,11 @@ ponder.on("CandidateConfirmer:block", async ({ event, context }) => {
event.block.number,
);

await refreshTwapExecutedTotals(
await refreshTwapExecutionState(
context,
chainId,
orphanCandidates.map((candidate) => candidate.generatorId),
event.block.number,
);

const preflightKnown = preflightStatuses.size;
Expand Down Expand Up @@ -380,10 +381,15 @@ ponder.on("CandidateConfirmer:block", async ({ event, context }) => {
}

if (confirmed > 0 || stale.length > 0) {
await refreshTwapExecutedTotals(context, chainId, [
...rowsToUpsert.map((row) => row.conditionalOrderGeneratorId),
...stale.map((candidate) => candidate.generatorId),
]);
await refreshTwapExecutionState(
context,
chainId,
[
...rowsToUpsert.map((row) => row.conditionalOrderGeneratorId),
...stale.map((candidate) => candidate.generatorId),
],
event.block.number,
);
log("info", "CandidateConfirmer:DONE", { block: String(event.block.number), chainId, candidates: unconfirmed.length, confirmed, expired: stale.length });
}
});
19 changes: 16 additions & 3 deletions src/application/handlers/block/orderStatusTracker.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ import {
import { fetchOrderStatusByUids } from "../../helpers/orderbookClient";
import { bumpGeneratorsUpdatedAt } from "../../helpers/updatedAtBlock";
import { log } from "../../helpers/logger";
import { refreshTwapExecutedTotals } from "../../helpers/executedAmounts";
import { refreshTwapExecutionState } from "../../helpers/executedAmounts";

const VALID_DISCRETE_STATUSES = new Set(["fulfilled", "unfilled", "expired", "cancelled"]);

Expand Down Expand Up @@ -108,10 +108,11 @@ ponder.on("OrderStatusTracker:block", async ({ event, context }) => {
event.block.number,
);

await refreshTwapExecutedTotals(
await refreshTwapExecutionState(
context,
chainId,
rowsToUpdate.map((row) => row.conditionalOrderGeneratorId),
event.block.number,
);

log("info", "OrderStatusTracker:DONE", { block: String(event.block.number), chainId, open: openOrders.length, updated: rowsToUpdate.length });
Expand Down Expand Up @@ -249,7 +250,12 @@ ponder.on("OrderStatusTracker:block", async ({ event, context }) => {

if (touchedGeneratorIds.length > 0) {
await bumpGeneratorsUpdatedAt(context, chainId, touchedGeneratorIds, event.block.number);
await refreshTwapExecutedTotals(context, chainId, touchedGeneratorIds);
await refreshTwapExecutionState(
context,
chainId,
touchedGeneratorIds,
event.block.number,
);
log("info", "OrderStatusTracker:REORG_HEAL", {
block: String(event.block.number),
chainId,
Expand Down Expand Up @@ -308,4 +314,11 @@ ponder.on("OrderStatusTracker:block", async ({ event, context }) => {
expired.map((r) => r.generatorId),
event.block.number,
);

await refreshTwapExecutionState(
context,
chainId,
expired.map((r) => r.generatorId),
event.block.number,
);
});
84 changes: 74 additions & 10 deletions src/application/helpers/executedAmounts.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import { and, eq, inArray, sql } from "ponder";
import {
candidateDiscreteOrder,
conditionalOrderGenerator,
discreteOrder,
type TwapAdditionalData,
Expand All @@ -12,36 +13,44 @@ const ZERO_TOTALS: TwapAdditionalData = {
executedFee: "0",
};

/** Rebuild TWAP parents' execution totals (additionalData) after part-order writes.
/** Rebuild TWAP parents' execution state after part-order writes.
* TWAP-only: every part sells the same token, so summing raw amounts is unit-safe
* (the orderbook reports executedFee in the sell token for sell orders). Other
* order types keep additionalData null — e.g. PerpetualSwap parts alternate
* direction, so a single sum would mix token units. */
export async function refreshTwapExecutedTotals(
export async function refreshTwapExecutionState(
context: Context,
chainId: number,
generatorIds: string[],
): Promise<void> {
blockNumber: bigint,
): Promise<string[]> {
const ids = [...new Set(generatorIds)];
if (ids.length === 0) return;
if (ids.length === 0) return [];

const generators = (await context.db.sql
.select({
eventId: conditionalOrderGenerator.eventId,
orderType: conditionalOrderGenerator.orderType,
status: conditionalOrderGenerator.status,
allCandidatesKnown: conditionalOrderGenerator.allCandidatesKnown,
})
.from(conditionalOrderGenerator)
.where(
and(
eq(conditionalOrderGenerator.chainId, chainId),
inArray(conditionalOrderGenerator.eventId, ids),
),
)) as { eventId: string; orderType: string }[];
)) as {
eventId: string;
orderType: string;
status: string;
allCandidatesKnown: boolean;
}[];

const twapIds = generators
.filter((generator) => generator.orderType === "TWAP")
.map((generator) => generator.eventId);
if (twapIds.length === 0) return;
if (twapIds.length === 0) return [];

// The .as() aliases are load-bearing: drizzle does not auto-alias raw sql
// fragments, so without them all three columns come back named "coalesce".
Expand All @@ -54,6 +63,8 @@ export async function refreshTwapExecutedTotals(
executedSellAmount: sql<string>`coalesce(sum(${discreteOrder.executedSellAmount}), 0)::text`.as("executed_sell_amount_sum"),
executedBuyAmount: sql<string>`coalesce(sum(${discreteOrder.executedBuyAmount}), 0)::text`.as("executed_buy_amount_sum"),
executedFee: sql<string>`coalesce(sum(${discreteOrder.executedFee}), 0)::text`.as("executed_fee_sum"),
partCount: sql<number>`count(*)::int`.as("part_count"),
openPartCount: sql<number>`count(*) filter (where ${discreteOrder.status} = 'open')::int`.as("open_part_count"),
})
.from(discreteOrder)
.where(
Expand All @@ -65,12 +76,65 @@ export async function refreshTwapExecutedTotals(
.groupBy(discreteOrder.conditionalOrderGeneratorId);

const totalsByGenerator = new Map(
rows.map(({ generatorId, ...totals }) => [generatorId, totals]),
rows.map(({ generatorId, partCount, openPartCount, ...totals }) => [
generatorId,
{ totals, partCount, openPartCount },
]),
);

for (const eventId of twapIds) {
const candidateGeneratorIds = new Set(
(
await context.db.sql
.select({
generatorId: candidateDiscreteOrder.conditionalOrderGeneratorId,
})
.from(candidateDiscreteOrder)
.where(
and(
eq(candidateDiscreteOrder.chainId, chainId),
inArray(candidateDiscreteOrder.conditionalOrderGeneratorId, twapIds),
),
)
.groupBy(candidateDiscreteOrder.conditionalOrderGeneratorId)
).map((row) => row.generatorId),
);

const completed: string[] = [];
for (const generator of generators) {
if (generator.orderType !== "TWAP") continue;

const aggregate = totalsByGenerator.get(generator.eventId);
const isComplete =
generator.status === "Active" &&
generator.allCandidatesKnown &&
aggregate != null &&
aggregate.partCount > 0 &&
aggregate.openPartCount === 0 &&
Comment on lines +107 to +112

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If I understood correctly, a child can be changed back to open on reorgs. In that case, I don't see any handler that will change the parent back to incomplete.

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks, fixed in f4a27a7

!candidateGeneratorIds.has(generator.eventId);
// Orderbook reorg reconciliation can reopen a previously terminal part.
const isReopened =
generator.status === "Completed" &&
((aggregate?.openPartCount ?? 0) > 0 ||
candidateGeneratorIds.has(generator.eventId));

await context.db

@Danziger Danziger Sep 8, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It would be better to collect the individual updates and send them together using db.batch: https://orm.drizzle.team/docs/batch-api, rather than awaiting inside the loop.

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There is no batch on this object https://ponder.sh/docs/indexing/write#store-api

image

.update(conditionalOrderGenerator, { chainId, eventId })
.set({ additionalData: totalsByGenerator.get(eventId) ?? ZERO_TOTALS });
.update(conditionalOrderGenerator, { chainId, eventId: generator.eventId })
.set({
additionalData: aggregate?.totals ?? ZERO_TOTALS,
...(isComplete && {
status: "Completed" as const,
lastPollResult: "executionState:allTerminal",
updatedAtBlock: blockNumber,
}),
...(isReopened && {
status: "Active" as const,
lastPollResult: "executionState:reopened",
updatedAtBlock: blockNumber,
}),
});

if (isComplete) completed.push(generator.eventId);
}

return completed;
}
9 changes: 7 additions & 2 deletions src/application/helpers/orderbook/client.ts
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@ import {
import { TimeoutError, withTimeout } from "../withTimeout";
import { bumpGeneratorsUpdatedAt } from "../updatedAtBlock";
import { log } from "../logger";
import { refreshTwapExecutedTotals } from "../executedAmounts";
import { refreshTwapExecutionState } from "../executedAmounts";
import { fetchAccountOrders, fetchOrdersByUids } from "./http";
import {
advanceOwnerOffset,
Expand Down Expand Up @@ -306,7 +306,12 @@ export async function upsertDiscreteOrders(
}

await bumpGeneratorsUpdatedAt(context, chainId, changedGeneratorIds, blockNumber);
await refreshTwapExecutedTotals(context, chainId, changedGeneratorIds);
await refreshTwapExecutionState(
context,
chainId,
changedGeneratorIds,
blockNumber,
);
return changedCount;
}

Expand Down
9 changes: 7 additions & 2 deletions src/application/helpers/uidPrecompute.ts
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ import { fetchOrderStatusByUids } from "./orderbookClient";
import { type OrderType, DETERMINISTIC_ORDER_TYPE } from "../../utils/order-types";
import { log } from "./logger";
import { MAX_TWAP_PRECOMPUTE_PARTS } from "../../constants";
import { refreshTwapExecutedTotals } from "./executedAmounts";
import { refreshTwapExecutionState } from "./executedAmounts";

// ─── Types ───────────────────────────────────────────────────────────────────

Expand Down Expand Up @@ -178,7 +178,12 @@ export async function precomputeAndDiscover(
}

if (discreteRows.length > 0) {
await refreshTwapExecutedTotals(context, chainId, [generatorEventId]);
await refreshTwapExecutionState(
context,
chainId,
[generatorEventId],
blockNumber,
);
}

const allTerminal = precomputed.every((o) => {
Expand Down
Loading
Loading