mirrors/pikasTech-unidesk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: add low-risk disk anti-bloat policy

Browse Source
This commit is contained in:
Codex
2026-05-28 18:06:51 +00:00
parent c9ffeaf775
commit aadc2646a3
8 changed files with 237 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/components/microservices/oa-event-flow/src/index.ts
+4 -11
View File
@@ -598,7 +598,7 @@ async function fetchTraceStats(executor: SqlExecutor, scopeId: string): Promise<
return rows[0] === undefined ? null : statsRowToRecord(rows[0]);
}
async function insertDerivedStatsEvent(executor: SqlExecutor, stats: TraceStatsRecord, sourceEvent: OaEventRecord): Promise<OaEventRecord | null> {
function traceStatsEvent(stats: TraceStatsRecord, sourceEvent: OaEventRecord): OaEventRecord {
const tags = ["stats", "trace", `service:${stats.serviceId}`, stats.scopeId];
if (typeof stats.taskId === "string" && stats.taskId.length > 0) addTag(tags, `task:${stats.taskId}`);
if (Number(stats.attemptIndex) > 0) addTag(tags, `attempt:${Number(stats.attemptIndex)}`);
@@ -611,7 +611,7 @@ async function insertDerivedStatsEvent(executor: SqlExecutor, stats: TraceStatsR
attemptIndex: Number(stats.attemptIndex) > 0 ? Number(stats.attemptIndex) : null,
stats,
};
const derived = normalizeEvent({
const event = normalizeEvent({
eventId: `oa-event-flow:trace-stats-updated:${stats.scopeId}:${stats.statsRevision}`,
type: "trace-stats-updated",
sourceKind: "projection",
@@ -623,13 +623,7 @@ async function insertDerivedStatsEvent(executor: SqlExecutor, stats: TraceStatsR
tags,
payload,
});
const rows = await executor<OaEventRow[]>`
INSERT INTO oa_events (event_id, type, source_kind, source_id, aggregate_type, aggregate_id, correlation_id, causation_id, tags, payload, created_at)
VALUES (${derived.eventId}, ${derived.type}, ${derived.sourceKind}, ${derived.sourceId}, ${derived.aggregateType}, ${derived.aggregateId}, ${derived.correlationId}, ${derived.causationId}, ${executor.json(derived.tags)}, ${executor.json(derived.payload)}, ${derived.createdAt})
ON CONFLICT (event_id) DO NOTHING
RETURNING *
`;
return rows[0] === undefined ? null : rowToEvent(rows[0]);
return { ...event, sequence: sourceEvent.sequence, createdAt: sourceEvent.createdAt };
}
async function applyTraceSnapshot(executor: SqlExecutor, event: OaEventRecord, scope: TraceScope): Promise<TraceStatsRecord | null> {
@@ -804,8 +798,7 @@ async function publishEvents(inputs: OaEventInput[]): Promise<{ inserted: OaEven
}
inserted.push(stored);
for (const stats of await applyTraceProjection(tx, stored)) {
const statsEvent = await insertDerivedStatsEvent(tx, stats, stored);
if (statsEvent !== null) derived.push(statsEvent);
derived.push(traceStatsEvent(stats, stored));
}
}
const lastSequence = Math.max(0, ...inserted.map((event) => event.sequence), ...derived.map((event) => event.sequence));