Skip to content

Commit c484be3

Browse files
DeFiVCDeFiVC
authored
Merge pull request #261 from christianahtemitope2005/fix/issues-224-225-226-227
fix: scope idempotency by user/endpoint, audit retry, HTTP tracing, reward helper extraction
2 parents f5060e9 + 540e340 commit c484be3

5 files changed

Lines changed: 117 additions & 128 deletions

File tree

‎package.json‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
"@opentelemetry/api": "^1.9.1",
2525
"@opentelemetry/exporter-trace-otlp-http": "^0.219.0",
2626
"@opentelemetry/instrumentation-fastify": "^0.56.0",
27+
"@opentelemetry/instrumentation-http": "^0.219.0",
2728
"@opentelemetry/instrumentation-ioredis": "^0.67.0",
2829
"@opentelemetry/instrumentation-pg": "^0.71.0",
2930
"@opentelemetry/resources": "^2.8.0",

‎src/audit/index.ts‎

Lines changed: 16 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,21 @@ interface AuditFields {
2626
userAgent?: string;
2727
}
2828

29-
export function auditLog(event: AuditEvent, fields: AuditFields): void {
29+
export async function auditLog(event: AuditEvent, fields: AuditFields): Promise<void> {
3030
logger.info({ audit: true, event, ...fields }, `audit: ${event}`);
31-
db.insert(auditLogs).values({ event, fields }).catch((err) => logger.error({ err }, "Failed to persist audit log"));
31+
for (let attempt = 0; attempt < 3; attempt++) {
32+
try {
33+
await db.insert(auditLogs).values({ event, fields });
34+
return;
35+
} catch (err) {
36+
if (attempt < 2) {
37+
await new Promise((resolve) => setTimeout(resolve, 100 * (attempt + 1)));
38+
continue;
39+
}
40+
logger.error({ err }, "Failed to persist audit log after 3 attempts");
41+
process.stdout.write(
42+
JSON.stringify({ audit: true, event, ...fields, persistError: String(err) }) + "\n",
43+
);
44+
}
45+
}
3246
}

‎src/middleware/idempotency.ts‎

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -12,8 +12,15 @@ export async function checkIdempotency(
1212
): Promise<{ cached: boolean; response?: { status: number; body: unknown } }> {
1313
const requestHash = sha256Hash(JSON.stringify(requestBody));
1414

15+
// Scope lookup to userId AND endpoint so the same key value used by two
16+
// different users (or at two different endpoints) never returns a cached
17+
// response from the wrong context — closes #227.
1518
const existing = await db.query.idempotencyKeys.findFirst({
16-
where: eq(idempotencyKeys.key, key),
19+
where: and(
20+
eq(idempotencyKeys.key, key),
21+
eq(idempotencyKeys.userId, userId),
22+
eq(idempotencyKeys.endpoint, endpoint),
23+
),
1724
});
1825

1926
if (existing) {
@@ -51,7 +58,11 @@ export async function checkIdempotency(
5158

5259
if (isUniqueViolation) {
5360
const concurrentRecord = await db.query.idempotencyKeys.findFirst({
54-
where: eq(idempotencyKeys.key, key),
61+
where: and(
62+
eq(idempotencyKeys.key, key),
63+
eq(idempotencyKeys.userId, userId),
64+
eq(idempotencyKeys.endpoint, endpoint),
65+
),
5566
});
5667

5768
if (

‎src/modules/rewards/reward.service.ts‎

Lines changed: 85 additions & 124 deletions
Original file line numberDiff line numberDiff line change
@@ -77,6 +77,70 @@ async function handleBadSeqError(submissionId: string, stellarAddress: string):
7777
return "pending_indexer_confirmation";
7878
}
7979

80+
interface RewardClaimData {
81+
submissionId: string;
82+
userId: string;
83+
score: number;
84+
stellarAddress: string;
85+
quizId: string;
86+
}
87+
88+
// Phase 2: invoke Stellar, observe metrics, handle bad_seq. Throws on all other errors.
89+
async function _executeStellarRewardClaim(claimData: RewardClaimData): Promise<string> {
90+
const proof = createQuizProof(claimData.userId, claimData.quizId, claimData.score);
91+
const txStart = process.hrtime.bigint();
92+
try {
93+
const txHash = await invokeContract(
94+
config.STELLAR_REWARD_CONTRACT_ID,
95+
"claim_reward",
96+
[
97+
StellarSdk.Address.fromString(claimData.stellarAddress).toScVal(),
98+
StellarSdk.nativeToScVal(claimData.score, { type: "u32" }),
99+
StellarSdk.nativeToScVal(Buffer.from(proof.signature, "base64")),
100+
],
101+
);
102+
stellarTxDurationSeconds.observe(
103+
{ method: "claim_reward", status: "success" },
104+
Number(process.hrtime.bigint() - txStart) / 1e9,
105+
);
106+
return txHash;
107+
} catch (err: unknown) {
108+
stellarTxDurationSeconds.observe(
109+
{ method: "claim_reward", status: "error" },
110+
Number(process.hrtime.bigint() - txStart) / 1e9,
111+
);
112+
if (
113+
err instanceof StellarError &&
114+
(err.message.includes("bad_seq") || err.message.includes("tx_bad_seq"))
115+
) {
116+
return handleBadSeqError(claimData.submissionId, claimData.stellarAddress);
117+
}
118+
throw err;
119+
}
120+
}
121+
122+
// Phase 3: write the confirmed or pending-confirmation outcome to the database.
123+
async function _applyRewardToDb(submissionId: string, userId: string, txHash: string): Promise<void> {
124+
const isPending = txHash === "pending_indexer_confirmation";
125+
await db.transaction(async (tx) => {
126+
if (isPending) {
127+
await tx
128+
.update(quizSubmissions)
129+
.set({ rewardClaimed: false, rewardPending: true, txHash })
130+
.where(eq(quizSubmissions.id, submissionId));
131+
} else {
132+
await tx
133+
.update(quizSubmissions)
134+
.set({ rewardClaimed: true, rewardPending: false, txHash, rewardAmount: REWARD_AMOUNT })
135+
.where(eq(quizSubmissions.id, submissionId));
136+
await tx
137+
.update(users)
138+
.set({ credits: sql`${users.credits} + ${REWARD_AMOUNT}` })
139+
.where(eq(users.id, userId));
140+
}
141+
});
142+
}
143+
80144
/**
81145
* Shared reward claim execution logic.
82146
* Used by both the direct claim path and the background retry processor.
@@ -140,70 +204,18 @@ export async function processRewardClaim(
140204
return true;
141205
}
142206

143-
// Phase 2: Execute Stellar transaction outside DB (no connection held)
144-
const proof = createQuizProof(userId, claimData.quizId, claimData.score);
145-
const txStart = process.hrtime.bigint();
207+
// Phase 2 & 3: Stellar invocation then DB update — shared with claimReward.
146208
let txHash: string;
147209
try {
148-
txHash = await invokeContract(
149-
config.STELLAR_REWARD_CONTRACT_ID,
150-
"claim_reward",
151-
[
152-
StellarSdk.Address.fromString(claimData.stellarAddress).toScVal(),
153-
StellarSdk.nativeToScVal(claimData.score, { type: "u32" }),
154-
StellarSdk.nativeToScVal(Buffer.from(proof.signature, "base64")),
155-
],
156-
);
157-
stellarTxDurationSeconds.observe(
158-
{ method: "claim_reward", status: "success" },
159-
Number(process.hrtime.bigint() - txStart) / 1e9,
160-
);
210+
txHash = await _executeStellarRewardClaim(claimData);
161211
} catch (err: unknown) {
162-
stellarTxDurationSeconds.observe(
163-
{ method: "claim_reward", status: "error" },
164-
Number(process.hrtime.bigint() - txStart) / 1e9,
165-
);
166-
if (err instanceof StellarError && (err.message.includes("bad_seq") || err.message.includes("tx_bad_seq"))) {
167-
txHash = await handleBadSeqError(submissionId, claimData.stellarAddress);
168-
} else {
169-
// Mark as failed and release pending status
170-
await db
171-
.update(quizSubmissions)
172-
.set({ rewardPending: false, rewardFailed: true })
173-
.where(eq(quizSubmissions.id, submissionId));
174-
throw err;
175-
}
212+
await db
213+
.update(quizSubmissions)
214+
.set({ rewardPending: false, rewardFailed: true })
215+
.where(eq(quizSubmissions.id, submissionId));
216+
throw err;
176217
}
177-
178-
// Phase 3: Update DB with result in a quick transaction.
179-
// isPending gates credit-granting on a confirmed on-chain result rather
180-
// than the uncertain bad_seq outcome above — see #104.
181-
const isPending = txHash === "pending_indexer_confirmation";
182-
await db.transaction(async (tx) => {
183-
if (isPending) {
184-
await tx
185-
.update(quizSubmissions)
186-
.set({ rewardClaimed: false, rewardPending: true, txHash })
187-
.where(eq(quizSubmissions.id, submissionId));
188-
} else {
189-
await tx
190-
.update(quizSubmissions)
191-
.set({
192-
rewardClaimed: true,
193-
rewardPending: false,
194-
txHash,
195-
rewardAmount: REWARD_AMOUNT,
196-
})
197-
.where(eq(quizSubmissions.id, submissionId));
198-
199-
await tx
200-
.update(users)
201-
.set({
202-
credits: sql`${users.credits} + ${REWARD_AMOUNT}`,
203-
})
204-
.where(eq(users.id, userId));
205-
}
206-
});
218+
await _applyRewardToDb(submissionId, userId, txHash);
207219

208220
return true;
209221
}).then(async (result) => {
@@ -295,24 +307,10 @@ export class RewardService {
295307
};
296308
});
297309

298-
// Phase 2: Execute Stellar transaction outside DB (no connection held)
299-
const proof = createQuizProof(
300-
userId,
301-
claimData.quizId,
302-
claimData.score,
303-
);
304-
305-
let txHash: string | null = null;
310+
// Phase 2 & 3: Stellar invocation then DB update — shared with processRewardClaim.
311+
let txHash: string;
306312
try {
307-
txHash = await invokeContract(
308-
config.STELLAR_REWARD_CONTRACT_ID,
309-
"claim_reward",
310-
[
311-
StellarSdk.Address.fromString(claimData.stellarAddress).toScVal(),
312-
StellarSdk.nativeToScVal(claimData.score, { type: "u32" }),
313-
StellarSdk.nativeToScVal(Buffer.from(proof.signature, "base64")),
314-
],
315-
);
313+
txHash = await _executeStellarRewardClaim(claimData);
316314
} catch (err: unknown) {
317315
if (err instanceof NotFoundError) throw err;
318316

@@ -321,15 +319,11 @@ export class RewardService {
321319
{ submissionId },
322320
"Stellar circuit breaker open — queuing reward for later",
323321
);
324-
// Release pending status before queuing
325322
await db
326323
.update(quizSubmissions)
327324
.set({ rewardPending: false })
328325
.where(eq(quizSubmissions.id, submissionId));
329-
await enqueueReward({
330-
submissionId,
331-
userId,
332-
});
326+
await enqueueReward({ submissionId, userId });
333327
rewardClaimsTotal.inc({ status: "queued" });
334328
auditLog("reward.queued", {
335329
userId,
@@ -342,54 +336,21 @@ export class RewardService {
342336
amount: REWARD_AMOUNT,
343337
txHash: null,
344338
queued: true,
345-
message:
346-
"Reward claim queued — Stellar is temporarily unavailable",
339+
message: "Reward claim queued — Stellar is temporarily unavailable",
347340
};
348341
}
349342

350-
if (err instanceof StellarError && (err.message.includes("bad_seq") || err.message.includes("tx_bad_seq"))) {
351-
txHash = await handleBadSeqError(submissionId, claimData.stellarAddress);
352-
} else {
353-
// Mark as failed and release pending status
354-
await db
355-
.update(quizSubmissions)
356-
.set({ rewardPending: false, rewardFailed: true })
357-
.where(eq(quizSubmissions.id, submissionId));
358-
logger.error({ err, submissionId }, "On-chain reward claim failed");
359-
throw new Error("Failed to process on-chain reward");
360-
}
343+
await db
344+
.update(quizSubmissions)
345+
.set({ rewardPending: false, rewardFailed: true })
346+
.where(eq(quizSubmissions.id, submissionId));
347+
logger.error({ err, submissionId }, "On-chain reward claim failed");
348+
throw new Error("Failed to process on-chain reward");
361349
}
362350

363-
// Phase 3: Update DB with result in a quick transaction.
364-
// isPending gates credit-granting on a confirmed on-chain result
365-
// rather than the uncertain bad_seq outcome above — see #104.
366-
const isPending = txHash === "pending_indexer_confirmation";
367-
await db.transaction(async (tx) => {
368-
if (isPending) {
369-
await tx
370-
.update(quizSubmissions)
371-
.set({ rewardClaimed: false, rewardPending: true, txHash })
372-
.where(eq(quizSubmissions.id, submissionId));
373-
} else {
374-
await tx
375-
.update(quizSubmissions)
376-
.set({
377-
rewardClaimed: true,
378-
rewardPending: false,
379-
txHash,
380-
rewardAmount: REWARD_AMOUNT,
381-
})
382-
.where(eq(quizSubmissions.id, submissionId));
383-
384-
await tx
385-
.update(users)
386-
.set({
387-
credits: sql`${users.credits} + ${REWARD_AMOUNT}`,
388-
})
389-
.where(eq(users.id, userId));
390-
}
391-
});
351+
await _applyRewardToDb(submissionId, userId, txHash);
392352

353+
const isPending = txHash === "pending_indexer_confirmation";
393354
if (isPending) {
394355
rewardClaimsTotal.inc({ status: "pending" });
395356
auditLog("reward.pending_confirmation", {

‎src/tracing.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-http";
33
import { resourceFromAttributes } from "@opentelemetry/resources";
44
import { ATTR_SERVICE_NAME } from "@opentelemetry/semantic-conventions";
55
import { FastifyInstrumentation } from "@opentelemetry/instrumentation-fastify";
6+
import { HttpInstrumentation } from "@opentelemetry/instrumentation-http";
67
import { PgInstrumentation } from "@opentelemetry/instrumentation-pg";
78
import { IORedisInstrumentation } from "@opentelemetry/instrumentation-ioredis";
89

@@ -21,6 +22,7 @@ export function initTracing(serviceName = "chainlearn-api"): void {
2122
traceExporter: new OTLPTraceExporter({ url: endpoint }),
2223
instrumentations: [
2324
new FastifyInstrumentation(),
25+
new HttpInstrumentation(),
2426
new PgInstrumentation(),
2527
new IORedisInstrumentation(),
2628
],

0 commit comments

Comments
 (0)