fix: bound hourly step awaits by the per-step wait limit (#56)
This commit is contained in:
@@ -28,6 +28,9 @@ const WORK_POLL_MAX_MS = 5_000;
|
||||
// ponytail: fixed 15-minute step wait bound (900_000 ms), no config knob;
|
||||
// make it configurable if slow steps become regular.
|
||||
const WORK_WAIT_BOUND_MS = 900_000;
|
||||
// An await that outlived WORK_WAIT_BOUND_MS. The underlying call keeps
|
||||
// running; its late durable result is fenced like any other late publication.
|
||||
const BOUND_EXPIRED = Symbol('hourly-bound-expired');
|
||||
const FIRST_DAY_OF_MONTH = 1;
|
||||
const NO_MILLISECONDS = 0;
|
||||
|
||||
@@ -155,7 +158,10 @@ export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit {
|
||||
);
|
||||
await this.waitWorks(userId, children.map((child) => child.id));
|
||||
for (const account of accounts) {
|
||||
await this.events.get(userId, account.accountId).catch(() => undefined);
|
||||
const pending = this.events
|
||||
.get(userId, account.accountId)
|
||||
.catch(() => undefined);
|
||||
await this.withBound(pending);
|
||||
}
|
||||
await this.settle(attempt, null, dispatchProblems);
|
||||
} catch {
|
||||
@@ -173,11 +179,14 @@ export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit {
|
||||
dispatchProblems: HourlySection[],
|
||||
) {
|
||||
const { userId, revision } = attempt;
|
||||
const member = await this.membership.request(userId, revision, true);
|
||||
if (member) {
|
||||
const member = await this.withBound(
|
||||
this.membership.request(userId, revision, true),
|
||||
);
|
||||
if (member && member !== BOUND_EXPIRED) {
|
||||
await dispatch({ id: member.id, section: 'membership' });
|
||||
await member.done;
|
||||
const done = await this.withBound(member.done);
|
||||
if (
|
||||
done !== BOUND_EXPIRED &&
|
||||
(await this.repository.workError(userId, member.id)) ===
|
||||
TbankErrorCode.InvalidToken
|
||||
) {
|
||||
@@ -185,19 +194,45 @@ export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit {
|
||||
return null;
|
||||
}
|
||||
} else {
|
||||
// No durable step id (rejected request or an expired wait): record the
|
||||
// observable problem; a late row stays fenced by its durable attempt id.
|
||||
dispatchProblems.push({
|
||||
section: 'membership',
|
||||
errorCode: TbankErrorCode.Unavailable,
|
||||
});
|
||||
}
|
||||
const current = await this.access.local(userId);
|
||||
if (current.revision !== revision) {
|
||||
const current = await this.withBound(this.access.local(userId));
|
||||
if (current === BOUND_EXPIRED || current.revision !== revision) {
|
||||
await this.settle(attempt, HOURLY_CAUSE_INTERRUPTED);
|
||||
return null;
|
||||
}
|
||||
return current.accounts;
|
||||
}
|
||||
|
||||
// Races a step await against the same per-step bound the row wait uses, so
|
||||
// a dispatch that never resolves cannot hold the attempt running forever.
|
||||
// The call itself is not cancelled; a rejection landing after the expiry is
|
||||
// swallowed here because the expiry already accounts the step as a problem.
|
||||
private withBound<T>(step: Promise<T>): Promise<T | typeof BOUND_EXPIRED> {
|
||||
step.catch(() => undefined);
|
||||
return new Promise<T | typeof BOUND_EXPIRED>((resolve, reject) => {
|
||||
const timer = setTimeout(
|
||||
() => resolve(BOUND_EXPIRED),
|
||||
WORK_WAIT_BOUND_MS,
|
||||
);
|
||||
step.then(
|
||||
(value) => {
|
||||
clearTimeout(timer);
|
||||
resolve(value);
|
||||
},
|
||||
(error: unknown) => {
|
||||
clearTimeout(timer);
|
||||
reject(error instanceof Error ? error : new Error(String(error)));
|
||||
},
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
private async refreshAccounts(
|
||||
userId: string,
|
||||
revision: string,
|
||||
@@ -241,8 +276,10 @@ export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit {
|
||||
dispatchProblems: HourlySection[],
|
||||
) {
|
||||
try {
|
||||
const work = await this.portfolio.request(userId, revision, accountId, true);
|
||||
if (work) {
|
||||
const work = await this.withBound(
|
||||
this.portfolio.request(userId, revision, accountId, true),
|
||||
);
|
||||
if (work && work !== BOUND_EXPIRED) {
|
||||
await dispatch({ accountId, id: work.id, section: 'portfolio' });
|
||||
return;
|
||||
}
|
||||
@@ -264,27 +301,33 @@ export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit {
|
||||
dispatchProblems: HourlySection[],
|
||||
) {
|
||||
try {
|
||||
const id = await this.operations.request(
|
||||
userId,
|
||||
[
|
||||
{
|
||||
accountId,
|
||||
from: monthStartOf(new Date()),
|
||||
to: new Date(),
|
||||
},
|
||||
],
|
||||
true,
|
||||
'screen',
|
||||
revision,
|
||||
const id = await this.withBound(
|
||||
this.operations.request(
|
||||
userId,
|
||||
[
|
||||
{
|
||||
accountId,
|
||||
from: monthStartOf(new Date()),
|
||||
to: new Date(),
|
||||
},
|
||||
],
|
||||
true,
|
||||
'screen',
|
||||
revision,
|
||||
),
|
||||
);
|
||||
if (id) await dispatch({ accountId, id, section: 'operations' });
|
||||
if (id && id !== BOUND_EXPIRED) {
|
||||
await dispatch({ accountId, id, section: 'operations' });
|
||||
return;
|
||||
}
|
||||
} catch {
|
||||
dispatchProblems.push({
|
||||
accountId,
|
||||
errorCode: TbankErrorCode.Unavailable,
|
||||
section: 'operations',
|
||||
});
|
||||
// A failed dispatch has no durable step row; record the problem.
|
||||
}
|
||||
dispatchProblems.push({
|
||||
accountId,
|
||||
errorCode: TbankErrorCode.Unavailable,
|
||||
section: 'operations',
|
||||
});
|
||||
}
|
||||
|
||||
private async dispatchEvents(
|
||||
@@ -294,17 +337,21 @@ export class TbankHourlyScheduler implements OnModuleDestroy, OnModuleInit {
|
||||
dispatchProblems: HourlySection[],
|
||||
) {
|
||||
try {
|
||||
const ids = await this.events.refresh(userId, accountId);
|
||||
for (const id of ids) {
|
||||
await dispatch({ accountId, id, section: 'events' });
|
||||
const ids = await this.withBound(this.events.refresh(userId, accountId));
|
||||
if (ids !== BOUND_EXPIRED) {
|
||||
for (const id of ids) {
|
||||
await dispatch({ accountId, id, section: 'events' });
|
||||
}
|
||||
return;
|
||||
}
|
||||
} catch {
|
||||
dispatchProblems.push({
|
||||
accountId,
|
||||
errorCode: TbankErrorCode.Unavailable,
|
||||
section: 'events',
|
||||
});
|
||||
// A failed dispatch has no durable step row; record the problem.
|
||||
}
|
||||
dispatchProblems.push({
|
||||
accountId,
|
||||
errorCode: TbankErrorCode.Unavailable,
|
||||
section: 'events',
|
||||
});
|
||||
}
|
||||
|
||||
private async settle(
|
||||
|
||||
Reference in New Issue
Block a user