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

fix: parallelize node trigger post sync

Browse Source
This commit is contained in:
Codex
2026-06-21 14:39:19 +00:00
parent d1ed8a54c0
commit b213eaca2b
1 changed files with 79 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/src/hwlab-node-impl.ts
+79 -1
View File
@@ -2678,6 +2678,7 @@ function nodeRuntimeTriggerCurrent(scoped: ReturnType<typeof parseNodeScopedDele
if (!scoped.rerun && before.exists === true && (before.status === "True" || before.status === "Unknown")) {
const pipelineWait = before.status === "Unknown"
? waitForNodeRuntimePipelineRunTerminal(spec, pipelineRun, scoped.timeoutSeconds, {
opportunisticPostSync: () => nodeRuntimeOpportunisticGitMirrorSync(scoped, sourceCommit, pipelineRun),
opportunisticPostFlush: () => nodeRuntimeOpportunisticGitMirrorFlush(scoped, sourceCommit, pipelineRun),
})
: { ok: true, status: "already-succeeded", pipelineRun: before, polls: 0, elapsedMs: 0 };
@@ -2769,6 +2770,7 @@ function nodeRuntimeTriggerCurrent(scoped: ReturnType<typeof parseNodeScopedDele
printNodeRuntimeTriggerProgress(spec, { stage: "create-pipelinerun", status: createOk ? "succeeded" : "failed", sourceCommit, pipelineRun, exitCode: create.exitCode, createObservedAfterTimeout: createObserved && !isCommandSuccess(create) ? true : undefined });
const pipelineWait = createOk
? waitForNodeRuntimePipelineRunTerminal(spec, pipelineRun, scoped.timeoutSeconds, {
opportunisticPostSync: () => nodeRuntimeOpportunisticGitMirrorSync(scoped, sourceCommit, pipelineRun),
opportunisticPostFlush: () => nodeRuntimeOpportunisticGitMirrorFlush(scoped, sourceCommit, pipelineRun),
})
: null;
@@ -3265,6 +3267,42 @@ function nodeRuntimeEnsureGitMirrorFlushed(
};
}
function nodeRuntimeOpportunisticGitMirrorSync(
scoped: ReturnType<typeof parseNodeScopedDelegatedOptions>,
sourceCommit: string,
pipelineRun: string,
): Record<string, unknown> | null {
const stage = "git-mirror-parallel-sync";
const full = nodeScopedFullOutput(scoped);
printNodeRuntimeTriggerProgress(scoped.spec, { stage, status: "started", sourceCommit, pipelineRun });
const sync = nodeRuntimeGitMirrorRun({ ...scoped, domain: "git-mirror", action: "sync", confirm: true, dryRun: false, wait: true });
const after = record(sync.status);
const afterSummary = Object.keys(after).length > 0 ? compactNodeRuntimeGitMirrorStatus(after) : {};
const ok = sync.ok === true;
printNodeRuntimeTriggerProgress(scoped.spec, {
stage,
status: ok ? "succeeded" : "failed",
sourceCommit,
pipelineRun,
jobName: sync.jobName ?? null,
...afterSummary,
});
return {
ok,
phase: "parallel",
mode: "synced-during-pipelinerun-wait",
executed: true,
sourceCommit,
pipelineRun,
sync: full ? sync : compactNodeRuntimeGitMirrorRun(sync),
jobName: sync.jobName ?? null,
after: full ? (Object.keys(after).length > 0 ? after : null) : undefined,
afterSummary: Object.keys(afterSummary).length > 0 ? afterSummary : null,
degradedReason: ok ? undefined : "node-runtime-git-mirror-parallel-sync-failed",
next: ok ? undefined : sync.next ?? { sync: `bun scripts/cli.ts hwlab nodes git-mirror sync --node ${scoped.node} --lane ${scoped.lane} --confirm --wait` },
};
}
function nodeRuntimeOpportunisticGitMirrorFlush(
scoped: ReturnType<typeof parseNodeScopedDelegatedOptions>,
sourceCommit: string,
@@ -5699,14 +5737,16 @@ function waitForNodeRuntimePipelineRunTerminal(
spec: HwlabRuntimeLaneSpec,
pipelineRun: string,
timeoutSeconds: number,
options: { opportunisticPostFlush?: () => Record<string, unknown> | null } = {},
options: { opportunisticPostFlush?: () => Record<string, unknown> | null; opportunisticPostSync?: () => Record<string, unknown> | null } = {},
): Record<string, unknown> {
const startedAt = Date.now();
const deadline = startedAt + timeoutSeconds * 1000;
let polls = 0;
let last: Record<string, unknown> = { exists: false, name: pipelineRun };
let lastOpportunisticPostFlushAt = 0;
let opportunisticPostSyncAttempted = false;
const opportunisticPostFlushes: Record<string, unknown>[] = [];
const opportunisticPostSyncs: Record<string, unknown>[] = [];
printNodeRuntimeTriggerProgress(spec, { stage: "pipelinerun-wait", status: "started", pipelineRun, timeoutSeconds });
while (Date.now() <= deadline) {
polls += 1;
@@ -5727,10 +5767,20 @@ function waitForNodeRuntimePipelineRunTerminal(
failureSummary,
polls,
elapsedMs: Date.now() - startedAt,
opportunisticPostSyncs: opportunisticPostSyncs.length > 0 ? opportunisticPostSyncs : undefined,
opportunisticPostFlushes: opportunisticPostFlushes.length > 0 ? opportunisticPostFlushes : undefined,
degradedReason: ok ? undefined : "node-runtime-pipelinerun-failed",
};
}
if (
options.opportunisticPostSync !== undefined
&& !opportunisticPostSyncAttempted
&& nodeRuntimePipelineRunSourceCloneCompleted(spec, pipelineRun)
) {
opportunisticPostSyncAttempted = true;
const sync = options.opportunisticPostSync();
if (sync !== null) opportunisticPostSyncs.push(sync);
}
if (options.opportunisticPostFlush !== undefined && Date.now() - lastOpportunisticPostFlushAt >= 15_000) {
lastOpportunisticPostFlushAt = Date.now();
const flush = options.opportunisticPostFlush();
@@ -5749,11 +5799,39 @@ function waitForNodeRuntimePipelineRunTerminal(
failureSummary,
polls,
elapsedMs: Date.now() - startedAt,
opportunisticPostSyncs: opportunisticPostSyncs.length > 0 ? opportunisticPostSyncs : undefined,
opportunisticPostFlushes: opportunisticPostFlushes.length > 0 ? opportunisticPostFlushes : undefined,
degradedReason: "node-runtime-pipelinerun-wait-timeout",
};
}
function nodeRuntimePipelineRunSourceCloneCompleted(spec: HwlabRuntimeLaneSpec, pipelineRun: string): boolean {
const result = runNodeK3sArgs(spec, ["kubectl", "-n", "hwlab-ci", "get", "taskrun", "-l", `tekton.dev/pipelineRun=${pipelineRun}`, "-o", "json"], 60);
if (result.exitCode !== 0) return false;
let parsed: unknown;
try {
parsed = JSON.parse(result.stdout);
} catch {
return false;
}
const items = Array.isArray(record(parsed).items) ? record(parsed).items as unknown[] : [];
return items.some((item) => {
const itemRecord = record(item);
const metadata = record(itemRecord.metadata);
const labels = record(metadata.labels);
const name = typeof metadata.name === "string" ? metadata.name : "";
const pipelineTask = typeof labels["tekton.dev/pipelineTask"] === "string" ? labels["tekton.dev/pipelineTask"] : "";
const taskName = `${pipelineTask} ${name}`;
if (!/(clone|checkout|local-git|source[-_ ]*worktree)/iu.test(taskName)) return false;
const status = record(itemRecord.status);
const conditions = Array.isArray(status.conditions) ? status.conditions as unknown[] : [];
return conditions.some((condition) => {
const conditionRecord = record(condition);
return conditionRecord.type === "Succeeded" && conditionRecord.status === "True";
});
});
}
function printNodeRuntimeTriggerProgress(spec: HwlabRuntimeLaneSpec, data: Record<string, unknown> = {}): void {
process.stderr.write(`${JSON.stringify({ event: "hwlab.runtime-lane.trigger.progress", at: new Date().toISOString(), lane: spec.lane, node: spec.nodeId, ...data })}\n`);
}