hectorzhao/lislgosms
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf: batch inbound workflows by tenant

Browse Source
This commit is contained in:
hectorzhao
2026-08-25 17:11:55 +08:00
parent d3ceeb1e16
commit 76e7c8c401
9 changed files with 184 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/src/send-chain/send-chain.service.spec.ts
+70 -1
View File
@@ -2449,11 +2449,80 @@ describe('SendChainService', () => {
const { service, prisma } = createService();
prisma.$queryRaw.mockResolvedValueOnce([]);
await (service as any).submission.inboundEntry.claimInboundWorkflows(5);
await (service as any).submission.inboundEntry.claimInboundWorkflows(5, ['tenant-busy']);
const sql = prisma.$queryRaw.mock.calls[0][0].strings.join(' ');
expect(sql).toContain("NOW() AT TIME ZONE 'UTC'");
expect(sql).toContain('"tenantId" NOT IN');
expect(prisma.$queryRaw.mock.calls[0][0].values).toContain('tenant-busy');
expect(sql).toContain('FOR UPDATE SKIP LOCKED');
expect(sql).toContain('inbox."tenantId"');
});
it('groups Inbox work by tenant and keeps chunks for one tenant serial', async () => {
const { service } = createService();
const inboundEntry = (service as any).submission.inboundEntry;
const items = [
{ id: 'a1', tenantId: 'tenant-a', applicationId: 'app-1' },
{ id: 'b1', tenantId: 'tenant-b', applicationId: 'app-2' },
{ id: 'a2', tenantId: 'tenant-a', applicationId: 'app-1' },
{ id: 'a3', tenantId: 'tenant-a', applicationId: 'app-1' },
];
const grouped = inboundEntry.groupInboundWorkflowsByTenant(items);
expect([...grouped.keys()]).toEqual(['tenant-a', 'tenant-b']);
expect(grouped.get('tenant-a').map((item: { id: string }) => item.id)).toEqual(['a1', 'a2', 'a3']);
inboundEntry.processClaimedInboundWorkflowBatch = jest.fn().mockResolvedValue(undefined);
await inboundEntry.processTenantInboundWorkflowBatches(grouped.get('tenant-a'), 2, new Map());
expect(inboundEntry.processClaimedInboundWorkflowBatch.mock.calls.map((call: unknown[]) => (
(call[0] as Array<{ id: string }>).map((item) => item.id)
))).toEqual([['a1', 'a2'], ['a3']]);
});
it('coalesces a small ready Inbox set before claiming the next tenant batch', async () => {
const previousWait = process.env.API_INBOUND_WORKFLOW_BATCH_WAIT_MS;
const previousTarget = process.env.API_INBOUND_WORKFLOW_TARGET_BATCH_SIZE;
process.env.API_INBOUND_WORKFLOW_BATCH_WAIT_MS = '5';
process.env.API_INBOUND_WORKFLOW_TARGET_BATCH_SIZE = '32';
try {
const { service, prisma } = createService();
prisma.$queryRaw.mockResolvedValueOnce([{ count: 3n }]);
const startedAt = Date.now();
await (service as any).submission.inboundEntry.waitForInboundWorkflowMicroBatch(96, ['tenant-busy']);
expect(Date.now() - startedAt).toBeGreaterThanOrEqual(4);
const query = prisma.$queryRaw.mock.calls[0][0];
expect(query.strings.join(' ')).toContain('COUNT(*)::bigint');
expect(query.values).toContain('tenant-busy');
} finally {
if (previousWait == null) delete process.env.API_INBOUND_WORKFLOW_BATCH_WAIT_MS;
else process.env.API_INBOUND_WORKFLOW_BATCH_WAIT_MS = previousWait;
if (previousTarget == null) delete process.env.API_INBOUND_WORKFLOW_TARGET_BATCH_SIZE;
else process.env.API_INBOUND_WORKFLOW_TARGET_BATCH_SIZE = previousTarget;
}
});
it('uses UTC for Submit Outbox claim, lease, publish, and retry timestamps', async () => {
const previousShadow = process.env.SEND_SUBMIT_OUTBOX_SHADOW_ENABLED;
const previousPublish = process.env.SEND_SUBMIT_OUTBOX_PUBLISH_ENABLED;
process.env.SEND_SUBMIT_OUTBOX_SHADOW_ENABLED = 'true';
delete process.env.SEND_SUBMIT_OUTBOX_PUBLISH_ENABLED;
try {
const { service, prisma } = createService();
prisma.$queryRaw.mockResolvedValueOnce([{ id: 'outbox-1', submitId: 'SUB-1', payload: { messageId: 'MSG-1' } }]);
const gatewaySubmit = (service as any).submission.gatewaySubmit;
await gatewaySubmit.publishSubmitOutboxBatch();
const claimSql = prisma.$queryRaw.mock.calls[0][0].strings.join(' ');
const publishSql = prisma.$executeRaw.mock.calls.at(-1)[0].strings.join(' ');
expect(claimSql).toContain("NOW() AT TIME ZONE 'UTC'");
expect(publishSql).toContain("NOW() AT TIME ZONE 'UTC'");
expect(`${claimSql} ${publishSql}`).not.toContain('CURRENT_TIMESTAMP');
} finally {
if (previousShadow == null) delete process.env.SEND_SUBMIT_OUTBOX_SHADOW_ENABLED;
else process.env.SEND_SUBMIT_OUTBOX_SHADOW_ENABLED = previousShadow;
if (previousPublish == null) delete process.env.SEND_SUBMIT_OUTBOX_PUBLISH_ENABLED;
else process.env.SEND_SUBMIT_OUTBOX_PUBLISH_ENABLED = previousPublish;
}
});
it('enqueues a freshly persisted inbound message without querying the task and message again', async () => {
api/src/send-chain/send-gateway-submit.service.ts
+11 -11
View File
@@ -298,7 +298,7 @@ startSubmitOutboxPublisher() {
"receiptStatus" = NULL,
"errorCode" = NULL,
"errorMessage" = NULL,
"updatedAt" = CURRENT_TIMESTAMP
"updatedAt" = (NOW() AT TIME ZONE 'UTC')
FROM (VALUES ${updates}) AS updates(id, "channelId", carrier, province, "submitId")
WHERE message.id = updates.id AND message.status = 'queued'
`);
@@ -432,7 +432,7 @@ startSubmitOutboxPublisher() {
const values = Prisma.join(failed.map(({ message, reason }) => Prisma.sql`(${message.id}::text, ${reason.slice(0, 1000)}::text)`));
await this.prisma.$executeRaw(Prisma.sql`
UPDATE "SmsMessageRecord" AS message
SET status = 'failed', "errorMessage" = failures.reason, "updatedAt" = CURRENT_TIMESTAMP
SET status = 'failed', "errorMessage" = failures.reason, "updatedAt" = (NOW() AT TIME ZONE 'UTC')
FROM (VALUES ${values}) AS failures(id, reason)
WHERE message.id = failures.id AND message.status = 'queued'
`);
@@ -728,8 +728,8 @@ startSubmitOutboxPublisher() {
SELECT id
FROM "GatewaySubmitOutbox"
WHERE (
(status = 'pending' AND "nextAttemptAt" <= CURRENT_TIMESTAMP)
OR (status = 'publishing' AND "leaseExpiresAt" < CURRENT_TIMESTAMP)
(status = 'pending' AND "nextAttemptAt" <= (NOW() AT TIME ZONE 'UTC'))
OR (status = 'publishing' AND "leaseExpiresAt" < (NOW() AT TIME ZONE 'UTC'))
)
ORDER BY "createdAt", id
LIMIT ${batchSize}
@@ -738,9 +738,9 @@ startSubmitOutboxPublisher() {
UPDATE "GatewaySubmitOutbox" AS outbox
SET status = 'publishing',
"leaseOwner" = ${this.submitOutboxLeaseOwner},
"leaseExpiresAt" = CURRENT_TIMESTAMP + (${leaseSeconds} * INTERVAL '1 second'),
"leaseExpiresAt" = (NOW() AT TIME ZONE 'UTC') + (${leaseSeconds} * INTERVAL '1 second'),
"attemptCount" = outbox."attemptCount" + 1,
"updatedAt" = CURRENT_TIMESTAMP
"updatedAt" = (NOW() AT TIME ZONE 'UTC')
FROM candidates
WHERE outbox.id = candidates.id
RETURNING outbox.id, outbox."submitId", outbox.payload
@@ -756,11 +756,11 @@ startSubmitOutboxPublisher() {
UPDATE "GatewaySubmitOutbox" AS outbox
SET status = 'published',
"streamEntryId" = published."streamEntryId",
"publishedAt" = CURRENT_TIMESTAMP,
"publishedAt" = (NOW() AT TIME ZONE 'UTC'),
"leaseOwner" = NULL,
"leaseExpiresAt" = NULL,
"lastError" = NULL,
"updatedAt" = CURRENT_TIMESTAMP
"updatedAt" = (NOW() AT TIME ZONE 'UTC')
FROM (VALUES ${values}) AS published(id, "streamEntryId")
WHERE outbox.id = published.id
AND outbox.status = 'publishing'
@@ -775,11 +775,11 @@ startSubmitOutboxPublisher() {
await this.prisma.$executeRaw(Prisma.sql`
UPDATE "GatewaySubmitOutbox"
SET status = CASE WHEN "attemptCount" >= 10 THEN 'dead' ELSE 'pending' END,
"nextAttemptAt" = CURRENT_TIMESTAMP + (LEAST(60, POWER(2, LEAST("attemptCount", 6))) * INTERVAL '1 second'),
"nextAttemptAt" = (NOW() AT TIME ZONE 'UTC') + (LEAST(60, POWER(2, LEAST("attemptCount", 6))) * INTERVAL '1 second'),
"leaseOwner" = NULL,
"leaseExpiresAt" = NULL,
"lastError" = ${message.slice(0, 1000)},
"updatedAt" = CURRENT_TIMESTAMP
"updatedAt" = (NOW() AT TIME ZONE 'UTC')
WHERE id = ${row.id} AND "leaseOwner" = ${this.submitOutboxLeaseOwner}
`);
} catch (recordError) {
@@ -1055,7 +1055,7 @@ async refreshTaskProgress(batchTaskId: string, knownSingleMessageStatus?: string
WHEN message.status IN ('submit_queued', 'submitted', 'unknown') THEN 'sending'
ELSE 'queued'
END,
"updatedAt" = CURRENT_TIMESTAMP
"updatedAt" = (NOW() AT TIME ZONE 'UTC')
FROM "SmsMessageRecord" AS message
WHERE task.id = ${batchTaskId}
AND task."sourceType" = 'cmpp'
api/src/send-chain/send-inbound-entry.service.ts
+71 -13
View File
@@ -28,6 +28,7 @@ type InboundWorkflowPayload = {
type ClaimedInboundWorkflow = {
id: string;
requestKey: string;
tenantId: string;
applicationId: string;
attempts: number;
payload: Prisma.JsonValue;
@@ -74,6 +75,7 @@ export class SendInboundEntryService {
private readonly logger = new Logger('SendChainService');
private readonly inboundWorkflowWorkerId = `${hostname()}:${process.pid}:${randomUUID()}`;
private readonly inboundWorkflowTasks = new Set<Promise<void>>();
private readonly inboundWorkflowActiveTenants = new Set<string>();
private inboundWorkflowTimer?: ReturnType<typeof setTimeout>;
private inboundWorkflowInFlight = 0;
private inboundWorkflowPumping = false;
@@ -1241,7 +1243,9 @@ startInboundWorkflowWorker() {
this.inboundWorkflowPumping = true;
try {
await this.refreshInboundWorkflowMetrics();
const claimed = await this.measureInboundStage('worker_claim', () => this.claimInboundWorkflows(available));
const activeTenantIds = [...this.inboundWorkflowActiveTenants];
await this.waitForInboundWorkflowMicroBatch(available, activeTenantIds);
const claimed = await this.measureInboundStage('worker_claim', () => this.claimInboundWorkflows(available, activeTenantIds));
const applications = claimed.length === 0
? []
: await this.prisma.smsApplication.findMany({
@@ -1250,14 +1254,15 @@ startInboundWorkflowWorker() {
});
const applicationById = new Map(applications.map((application) => [application.id, application]));
const batchSize = positiveInteger(process.env.API_INBOUND_WORKFLOW_BATCH_SIZE, 64);
for (let offset = 0; offset < claimed.length; offset += batchSize) {
const batch = claimed.slice(offset, offset + batchSize);
this.inboundWorkflowInFlight += batch.length;
for (const [tenantId, tenantItems] of this.groupInboundWorkflowsByTenant(claimed)) {
this.inboundWorkflowActiveTenants.add(tenantId);
this.inboundWorkflowInFlight += tenantItems.length;
this.metrics?.setInboundWorkflowSlots(concurrency, this.inboundWorkflowInFlight);
const task = this.processClaimedInboundWorkflowBatch(batch, applicationById)
const task = this.processTenantInboundWorkflowBatches(tenantItems, batchSize, applicationById)
.catch((error) => this.logger.error(`CMPP inbound workflow batch failed to settle: ${String(error)}`))
.finally(() => {
this.inboundWorkflowInFlight = Math.max(0, this.inboundWorkflowInFlight - batch.length);
this.inboundWorkflowActiveTenants.delete(tenantId);
this.inboundWorkflowInFlight = Math.max(0, this.inboundWorkflowInFlight - tenantItems.length);
this.metrics?.setInboundWorkflowSlots(concurrency, this.inboundWorkflowInFlight);
this.inboundWorkflowTasks.delete(task);
this.scheduleInboundWorkflowPump(0);
@@ -1278,19 +1283,72 @@ startInboundWorkflowWorker() {
}
}
private claimInboundWorkflows(limit: number) {
private groupInboundWorkflowsByTenant(items: ClaimedInboundWorkflow[]) {
const grouped = new Map<string, ClaimedInboundWorkflow[]>();
for (const item of items) {
const tenantItems = grouped.get(item.tenantId) ?? [];
tenantItems.push(item);
grouped.set(item.tenantId, tenantItems);
}
return grouped;
}
private async processTenantInboundWorkflowBatches(
items: ClaimedInboundWorkflow[],
batchSize: number,
applicationById: Map<string, InboundApplication>,
) {
// A tenant account is one financial consistency boundary. Chunks for the
// same tenant stay serial, while different tenants are started as separate
// tasks by the pump and can use the configured workflow concurrency.
for (let offset = 0; offset < items.length; offset += batchSize) {
await this.processClaimedInboundWorkflowBatch(items.slice(offset, offset + batchSize), applicationById);
}
}
private async waitForInboundWorkflowMicroBatch(limit: number, excludedTenantIds: string[]) {
const maxWaitMs = Math.min(250, getNonNegativeConfigInteger(process.env, 'API_INBOUND_WORKFLOW_BATCH_WAIT_MS', 40));
if (maxWaitMs === 0 || limit < 2) return;
const configuredBatchSize = positiveInteger(process.env.API_INBOUND_WORKFLOW_BATCH_SIZE, 64);
const targetBatchSize = Math.min(
limit,
configuredBatchSize,
positiveInteger(process.env.API_INBOUND_WORKFLOW_TARGET_BATCH_SIZE, 32),
);
if (targetBatchSize < 2) return;
const excludedFilter = excludedTenantIds.length === 0
? Prisma.empty
: Prisma.sql`AND "tenantId" NOT IN (${Prisma.join(excludedTenantIds)})`;
const rows = await this.prisma.$queryRaw<Array<{ count: bigint }>>(Prisma.sql`
SELECT COUNT(*)::bigint AS count
FROM "CmppInboundSubmissionInbox"
WHERE status = 'pending'
AND "nextAttemptAt" <= (NOW() AT TIME ZONE 'UTC')
${excludedFilter}
`);
const ready = Number(rows[0]?.count ?? 0);
if (ready > 0 && ready < targetBatchSize) await sleep(maxWaitMs);
}
private claimInboundWorkflows(limit: number, excludedTenantIds: string[] = []) {
const staleSeconds = positiveInteger(process.env.API_INBOUND_WORKFLOW_STALE_SECONDS, 300);
const excludedFilter = excludedTenantIds.length === 0
? Prisma.empty
: Prisma.sql`AND "tenantId" NOT IN (${Prisma.join(excludedTenantIds)})`;
return this.prisma.$queryRaw<ClaimedInboundWorkflow[]>(Prisma.sql`
WITH candidates AS (
SELECT id
FROM "CmppInboundSubmissionInbox"
WHERE (
status = 'pending'
AND "nextAttemptAt" <= (NOW() AT TIME ZONE 'UTC')
) OR (
status = 'processing'
AND "lockedAt" <= (NOW() AT TIME ZONE 'UTC') - make_interval(secs => ${staleSeconds})
(
status = 'pending'
AND "nextAttemptAt" <= (NOW() AT TIME ZONE 'UTC')
) OR (
status = 'processing'
AND "lockedAt" <= (NOW() AT TIME ZONE 'UTC') - make_interval(secs => ${staleSeconds})
)
)
${excludedFilter}
-- Priority applications enter the same durable Inbox, but are claimed first while
-- preserving FIFO within each class. This keeps the V5 priority contract effective
-- before BullMQ without adding another non-durable queue.
@@ -1306,7 +1364,7 @@ startInboundWorkflowWorker() {
"updatedAt" = (NOW() AT TIME ZONE 'UTC')
FROM candidates
WHERE inbox.id = candidates.id
RETURNING inbox.id, inbox."requestKey", inbox."applicationId", inbox.attempts, inbox.payload
RETURNING inbox.id, inbox."requestKey", inbox."tenantId", inbox."applicationId", inbox.attempts, inbox.payload
`);
}