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

fix(workbench): show and accelerate Kafka replay

Browse Source
This commit is contained in:
root
2026-07-20 04:45:15 +02:00
parent f3563f7327
commit 57af108d40
10 changed files with 107 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cloud/kafka-event-bridge.test.ts
+8 -1
View File
@@ -6,7 +6,7 @@ import { test } from "bun:test";
import { createCloudApiBunServer } from "./bun-server.ts";
import { buildCloudApiReadiness } from "./health-contract.ts";
import { decodeCanonicalAgentRunKafkaMessage, kafkaDnsLookup, kafkaEventBridgeConfig, kafkaPartitionForKey, projectAgentRunKafkaEventToHwlabEvent, publishAgentRunKafkaMessageLive, relayHwlabKafkaOutboxOnce, startHwlabKafkaEventBridge } from "./kafka-event-bridge.ts";
import { decodeCanonicalAgentRunKafkaMessage, kafkaDnsLookup, kafkaEventBridgeConfig, kafkaMessageKeyCanMatchPartitionKey, kafkaPartitionForKey, projectAgentRunKafkaEventToHwlabEvent, publishAgentRunKafkaMessageLive, relayHwlabKafkaOutboxOnce, startHwlabKafkaEventBridge } from "./kafka-event-bridge.ts";
test("Workbench refresh replay resolves the same Kafka partition as the session-keyed producer", () => {
const offsets = [
@@ -23,6 +23,13 @@ test("Workbench refresh replay resolves the same Kafka partition as the session-
assert.equal(kafkaPartitionForKey(null, "hwlab.event.v1", offsets), null);
});
test("Workbench refresh replay rejects foreign keyed values before JSON parsing", () => {
assert.equal(kafkaMessageKeyCanMatchPartitionKey("ses_target", "ses_target"), true);
assert.equal(kafkaMessageKeyCanMatchPartitionKey("ses_foreign", "ses_target"), false);
assert.equal(kafkaMessageKeyCanMatchPartitionKey(null, "ses_target"), true);
assert.equal(kafkaMessageKeyCanMatchPartitionKey("ses_target", null), true);
});
const PROJECTOR_ENV = Object.freeze({
HWLAB_KAFKA_DIRECT_PUBLISH_ENABLED: "false",
HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED: "false",
internal/cloud/kafka-event-bridge.ts
+31 -16
View File
@@ -973,6 +973,7 @@ export async function queryKafkaEventStream({ env = process.env, stream = "hwlab
const clientId = stringValue(env.HWLAB_KAFKA_CLIENT_ID) || DEFAULT_CLIENT_ID;
const resolvedTopic = stringValue(topic) || kafkaTopicForStream(stream, env);
const maxEvents = Math.max(1, integerValue(limit) || DEFAULT_QUERY_LIMIT);
const resolvedPartitionKey = stringValue(partitionKey);
const maxScannedRecords = scanLimit === null || scanLimit === undefined
? null
: strictPositiveInteger(scanLimit, "scanLimit");
@@ -989,7 +990,7 @@ export async function queryKafkaEventStream({ env = process.env, stream = "hwlab
timing.endOffsetSnapshotMs = Date.now() - stageStartedAtMs;
const endOffsetByPartition = new Map(endOffsetSnapshot.offsets.map((entry) => [entry.partition, entry.endOffset]));
const startOffsetByPartition = new Map(endOffsetSnapshot.offsets.map((entry) => [entry.partition, entry.startOffset]));
const targetPartition = kafkaPartitionForKey(partitionKey, resolvedTopic, endOffsetSnapshot.offsets);
const targetPartition = kafkaPartitionForKey(resolvedPartitionKey, resolvedTopic, endOffsetSnapshot.offsets);
const skippedPartitions = targetPartition === null ? [] : endOffsetSnapshot.offsets.filter((entry) => entry.partition !== targetPartition).map((entry) => entry.partition);
const requiresScopedGroupOffsets = skippedPartitions.length > 0;
const reachedEndPartitions = new Set([...skippedPartitions, ...endOffsetSnapshot.offsets.filter(kafkaOffsetPartitionIsEmpty).map((entry) => entry.partition)]);
@@ -1009,6 +1010,7 @@ export async function queryKafkaEventStream({ env = process.env, stream = "hwlab
let parsedCount = 0;
let invalidJsonCount = 0;
let filterRejectedCount = 0;
let keyRejectedCount = 0;
let excludedPostBarrierCount = 0;
let completionReason = "timeout";
let timer = null;
@@ -1090,22 +1092,28 @@ export async function queryKafkaEventStream({ env = process.env, stream = "hwlab
}
scannedCount += 1;
lastScannedOffsetByPartition.set(partition, stringValue(message.offset));
const valueText = message.value ? Buffer.from(message.value).toString("utf8") : "";
const value = parseJson(valueText);
if (!value || typeof value !== "object" || Array.isArray(value)) invalidJsonCount += 1;
else {
parsedCount += 1;
if (!eventMatchesFilters(value, { traceId, sessionId, runId, commandId })) filterRejectedCount += 1;
const messageKey = message.key ? Buffer.from(message.key).toString("utf8") : null;
if (!kafkaMessageKeyCanMatchPartitionKey(messageKey, resolvedPartitionKey)) {
filterRejectedCount += 1;
keyRejectedCount += 1;
} else {
const valueText = message.value ? Buffer.from(message.value).toString("utf8") : "";
const value = parseJson(valueText);
if (!value || typeof value !== "object" || Array.isArray(value)) invalidJsonCount += 1;
else {
events.push({
topic: messageTopic,
partition,
offset: message.offset,
key: message.key ? Buffer.from(message.key).toString("utf8") : null,
timestamp: message.timestamp ?? null,
valueSha256: sha256(valueText),
value
});
parsedCount += 1;
if (!eventMatchesFilters(value, { traceId, sessionId, runId, commandId })) filterRejectedCount += 1;
else {
events.push({
topic: messageTopic,
partition,
offset: message.offset,
key: messageKey,
timestamp: message.timestamp ?? null,
valueSha256: sha256(valueText),
value
});
}
}
}
if (fromBeginning !== false && endOffset && kafkaOffsetReached(message.offset, endOffset)) reachedEndPartitions.add(partition);
@@ -1153,6 +1161,7 @@ export async function queryKafkaEventStream({ env = process.env, stream = "hwlab
matchedCount: events.length,
invalidJsonCount,
filterRejectedCount,
keyRejectedCount,
excludedPostBarrierCount,
completionReason,
completion: {
@@ -1190,6 +1199,12 @@ export async function queryKafkaEventStream({ env = process.env, stream = "hwlab
}
}
export function kafkaMessageKeyCanMatchPartitionKey(messageKey, partitionKey) {
const expected = stringValue(partitionKey);
const actual = stringValue(messageKey);
return !expected || !actual || actual === expected;
}
export function kafkaPartitionForKey(partitionKey, topic, offsets = []) {
const key = stringValue(partitionKey);
const partitionMetadata = offsets
internal/cloud/server-workbench-realtime-http.ts
+3
View File
@@ -778,7 +778,10 @@ async function handleKafkaRefreshReplayWorkbenchRealtimeHttp(request, response,
completionReason: result?.completionReason ?? null,
complete: result?.completion?.complete === true,
scannedCount: result?.scannedCount ?? null,
parsedCount: result?.parsedCount ?? null,
matchedCount: result?.matchedCount ?? null,
filterRejectedCount: result?.filterRejectedCount ?? null,
keyRejectedCount: result?.keyRejectedCount ?? null,
partitionKeyScoped: result?.partitionKeyScoped === true,
targetPartition: result?.targetPartition ?? null,
timing: result?.timing ?? null,
internal/cloud/workbench-kafka-refresh-handoff.ts
+3
View File
@@ -403,7 +403,10 @@ function boundedQueryDiagnostics(result) {
completionReason: textValue(result?.completion?.reason ?? result?.completionReason),
complete: result?.completion?.complete === true,
scannedCount: numericValue(result?.scannedCount),
parsedCount: numericValue(result?.parsedCount),
matchedCount: numericValue(result?.matchedCount),
filterRejectedCount: numericValue(result?.filterRejectedCount),
keyRejectedCount: numericValue(result?.keyRejectedCount),
partitionKeyScoped: result?.partitionKeyScoped === true,
targetPartition: partitionValue(result?.targetPartition),
timing: {
internal/workbench/http.ts
+3
View File
@@ -170,7 +170,10 @@ function nativeKafkaRefreshEventStream(bridge: any, sessionId: string, traceId:
completionReason: result?.completionReason ?? null,
complete: result?.completion?.complete === true,
scannedCount: result?.scannedCount ?? null,
parsedCount: result?.parsedCount ?? null,
matchedCount: result?.matchedCount ?? null,
filterRejectedCount: result?.filterRejectedCount ?? null,
keyRejectedCount: result?.keyRejectedCount ?? null,
partitionKeyScoped: result?.partitionKeyScoped === true,
targetPartition: result?.targetPartition ?? null,
timing: result?.timing ?? null,
internal/workbench/workbench.test.ts
+6
View File
@@ -185,7 +185,10 @@ describe("Workbench native HTTP adapter", () => {
endOffsetsAvailable: true,
endOffsets: [{ partition: 0, startOffset: "0", endOffset: "100" }],
scannedCount: 100,
parsedCount: 100,
matchedCount: 100,
filterRejectedCount: 0,
keyRejectedCount: 0,
partitionKeyScoped: true,
targetPartition: 0,
timing: { endOffsetSnapshotMs: 2, groupOffsetsMs: 0, consumerConnectMs: 3, consumerSubscribeMs: 1, scanMs: 7, cleanupMs: 1, totalMs: 14 }
@@ -222,7 +225,10 @@ describe("Workbench native HTTP adapter", () => {
query: {
complete: true,
scannedCount: 100,
parsedCount: 100,
matchedCount: 100,
filterRejectedCount: 0,
keyRejectedCount: 0,
partitionKeyScoped: true,
targetPartition: 0,
timing: { scanMs: 7, totalMs: 14 }