Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 15 additions & 4 deletions packages/benchsdk-worker/src/reporter.ts
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,7 @@ export class BenchmarkReporter {
private pending: TaskResultRecord[] = [];
private sequenceNumber = 0;
private flushChain: Promise<void> = Promise.resolve();
private flushFailed = false;
private progress: BenchmarkReporterProgress;
private barrier: { step: string; concurrency: WorkerConcurrencySample[] } | null = null;

Expand Down Expand Up @@ -184,6 +185,7 @@ export class BenchmarkReporter {

flush(isFinal = false): Promise<void> {
this.flushChain = this.flushChain.then(async () => {
this.flushFailed = false;
while (this.pending.length >= this.cfg.batchSize || (isFinal && this.pending.length > 0)) {
const batch = this.pending.slice(0, this.cfg.batchSize);
try {
Expand All @@ -197,10 +199,9 @@ export class BenchmarkReporter {
records: batch,
});
} catch (error) {
// Results that never reach the platform are invisible otherwise: the
// worker still completes and the run just reports fewer tasks.
this.flushFailed = true;
console.warn(
`[benchsdk] dropping ${this.pending.length} unsent task result(s) for worker ${this.assignment.workerId}: ` +
`[benchsdk] failed to send ${this.pending.length} task result(s) for worker ${this.assignment.workerId}: ` +
`${error instanceof Error ? error.message : String(error)}`,
);
break;
Expand All @@ -213,7 +214,7 @@ export class BenchmarkReporter {
}

async finish(failed = false, error?: unknown): Promise<void> {
await this.flush(true).catch(() => {});
await this.flush(true);
if (failed) {
await this.client.failWorker(
this.cfg.benchmarkSlug,
Expand All @@ -224,6 +225,16 @@ export class BenchmarkReporter {
).catch(() => {});
return;
}
if (this.flushFailed || this.pending.length > 0) {
await this.client.failWorker(
this.cfg.benchmarkSlug,
this.cfg.runId,
this.assignment.workerId,
this.assignment.attemptId,
new Error('Failed to flush task results'),
).catch(() => {});
return;
}
await this.client.completeWorker(
this.cfg.benchmarkSlug,
this.cfg.runId,
Expand Down
Loading