fix: harden auth and investment synchronization
Backend security / security (push) Successful in 8s
Backend verify / verify (push) Successful in 11m9s
Conventional commits / commits (pull_request) Successful in 28s
Backend security / security (pull_request) Successful in 7s
Backend verify / verify (pull_request) Successful in 10m38s
Backend security / security (push) Successful in 8s
Backend verify / verify (push) Successful in 11m9s
Conventional commits / commits (pull_request) Successful in 28s
Backend security / security (pull_request) Successful in 7s
Backend verify / verify (pull_request) Successful in 10m38s
Resolve A01-A10 and C01-C07 from the backend audit. Preserve auth failures, forced candle reconciliation and forecast freshness. Validate MOEX responses and nested HTTP errors; remove unused legacy paths. Verified: 209 unit tests, 262 SQLite e2e tests on Linux, OpenAPI and durability. Record the remaining macOS native SQLite abort in the audit.
This commit is contained in:
@@ -13,8 +13,8 @@ import {
|
||||
TbankAccessService,
|
||||
TbankClient,
|
||||
type TbankShare,
|
||||
type TbankSyncAttempt,
|
||||
} from '../../../../integrations/tbank/tbank.api';
|
||||
import type { TbankSyncAttempt } from '../../../../integrations/tbank/tbank-access.service';
|
||||
import { investmentResourceMeta } from '../../investment-resource.mapper';
|
||||
import { CatalogEnrichmentService } from './catalog-enrichment.service';
|
||||
import { CandleRepository } from '../infrastructure/candle.repository';
|
||||
@@ -39,8 +39,12 @@ type Producer = {
|
||||
force: boolean;
|
||||
priority: number;
|
||||
observers: TbankSyncAttempt[];
|
||||
ranges?: CandleRange[];
|
||||
nextWindow?: number;
|
||||
pass?: {
|
||||
now: Date;
|
||||
force: boolean;
|
||||
ranges: CandleRange[];
|
||||
nextWindow: number;
|
||||
};
|
||||
};
|
||||
const months = { month: 1, halfYear: 6, year: 12 };
|
||||
const tailFreshForMs = 60_000;
|
||||
@@ -323,9 +327,14 @@ export class CandleService {
|
||||
id: reserved.work.id,
|
||||
attemptId: reserved.work.attemptId,
|
||||
};
|
||||
const producer = this.producers.get(scope(input));
|
||||
if (producer) {
|
||||
producer.force ||= force;
|
||||
producer.priority = Math.min(producer.priority, priority);
|
||||
if (force && now > producer.now) producer.now = now;
|
||||
}
|
||||
if (reserved.launch) {
|
||||
this.active.add(attempt.attemptId);
|
||||
const producer = this.producers.get(scope(input));
|
||||
if (producer) producer.observers.push(attempt);
|
||||
else {
|
||||
const next = { input, attempt, now, force, priority, observers: [] };
|
||||
@@ -346,7 +355,7 @@ export class CandleService {
|
||||
}
|
||||
|
||||
private async load(producer: Producer) {
|
||||
const { input, attempt, now } = producer;
|
||||
const { input, attempt } = producer;
|
||||
const instrument = await this.sources.ensureDetail(
|
||||
attempt.userId,
|
||||
attempt.revision,
|
||||
@@ -354,20 +363,24 @@ export class CandleService {
|
||||
input.uid,
|
||||
);
|
||||
if (!instrument) throw new NotFoundException();
|
||||
const range = rangeFor(input, instrument, now);
|
||||
if (!producer.ranges) {
|
||||
let pass = producer.pass;
|
||||
if (!pass || (producer.force && (!pass.force || producer.now > pass.now))) {
|
||||
const { now, force } = producer;
|
||||
const range = rangeFor(input, instrument, now);
|
||||
const saved = await this.repository.read(input.uid, range);
|
||||
producer.ranges = (
|
||||
producer.force ? [range] : candleGaps(range, saved.coverage, now)
|
||||
)
|
||||
.flatMap(candleWindows)
|
||||
.reverse();
|
||||
producer.nextWindow = 0;
|
||||
pass = producer.pass = {
|
||||
now,
|
||||
force,
|
||||
ranges: (force ? [range] : candleGaps(range, saved.coverage, now))
|
||||
.flatMap(candleWindows)
|
||||
.reverse(),
|
||||
nextWindow: 0,
|
||||
};
|
||||
}
|
||||
const ranges = producer.ranges;
|
||||
const index = producer.nextWindow ?? 0;
|
||||
const { ranges, now, force } = pass;
|
||||
const index = pass.nextWindow;
|
||||
let window = ranges[index];
|
||||
if (window && !producer.force) {
|
||||
if (window && !force) {
|
||||
const latest = await this.repository.read(input.uid, window);
|
||||
const gaps = candleGaps(window, latest.coverage, now)
|
||||
.flatMap(candleWindows)
|
||||
@@ -398,7 +411,7 @@ export class CandleService {
|
||||
if (!(await this.repository.stage(attempt, window, rows, now)))
|
||||
return false;
|
||||
if (
|
||||
!producer.force &&
|
||||
!force &&
|
||||
!(await this.repository.publish(
|
||||
attempt,
|
||||
{ uid: input.uid, ranges: [window] },
|
||||
@@ -406,13 +419,10 @@ export class CandleService {
|
||||
))
|
||||
)
|
||||
return false;
|
||||
producer.nextWindow = index + 1;
|
||||
if (producer.nextWindow < ranges.length) return null;
|
||||
pass.nextWindow = index + 1;
|
||||
if (pass.nextWindow < ranges.length) return null;
|
||||
}
|
||||
return this.repository.publish(attempt, {
|
||||
uid: input.uid,
|
||||
ranges: producer.force ? ranges : [],
|
||||
});
|
||||
return true;
|
||||
}
|
||||
|
||||
private async drain() {
|
||||
@@ -443,20 +453,43 @@ export class CandleService {
|
||||
code = response.code;
|
||||
}
|
||||
}
|
||||
if (!success) await this.repository.fail(producer.attempt, code);
|
||||
this.active.delete(producer.attempt.attemptId);
|
||||
for (const observer of producer.observers) {
|
||||
if (
|
||||
!success ||
|
||||
!(await this.repository.completeObserver(
|
||||
observer,
|
||||
producer.attempt,
|
||||
))
|
||||
)
|
||||
await this.repository.fail(observer, code);
|
||||
this.active.delete(observer.attemptId);
|
||||
// Seal publication with reservations so a late forced join cannot be
|
||||
// acknowledged after the final demand check but before work completion.
|
||||
const finished = this.reservations.then(async () => {
|
||||
const pass = producer.pass;
|
||||
if (success && pass) {
|
||||
if (producer.force && (!pass.force || producer.now > pass.now))
|
||||
return false;
|
||||
try {
|
||||
success = await this.repository.publish(producer.attempt, {
|
||||
uid: producer.input.uid,
|
||||
ranges: pass.force ? pass.ranges : [],
|
||||
});
|
||||
} catch {
|
||||
success = false;
|
||||
}
|
||||
}
|
||||
if (!success) await this.repository.fail(producer.attempt, code);
|
||||
this.active.delete(producer.attempt.attemptId);
|
||||
for (const observer of producer.observers) {
|
||||
if (
|
||||
!success ||
|
||||
!(await this.repository.completeObserver(
|
||||
observer,
|
||||
producer.attempt,
|
||||
))
|
||||
)
|
||||
await this.repository.fail(observer, code);
|
||||
this.active.delete(observer.attemptId);
|
||||
}
|
||||
this.producers.delete(scope(producer.input));
|
||||
return true;
|
||||
});
|
||||
this.reservations = finished.catch(() => undefined);
|
||||
if (!(await finished)) {
|
||||
this.queue.push(producer);
|
||||
continue;
|
||||
}
|
||||
this.producers.delete(scope(producer.input));
|
||||
if (success && !producer.force && producer.input.period !== 'all')
|
||||
await this.request(
|
||||
producer.attempt.userId,
|
||||
|
||||
Reference in New Issue
Block a user