mirror of
https://github.com/hanzoai/insights.git
synced 2026-10-30 16:15:31 +00:00
zeekayandHanzo Dev
859392dcdb
Two physical names were carried on the grounds that a table is a fact rather than a word. Both tables were empty, which is the only thing that ever made that argument hold. hog_invocation_results is invocations. It holds one row per function invocation, so it says so now. All four objects -- the distributed read alias, the local data table, the Kafka engine table and the MV between them -- were measured at zero rows against the live cluster before this was written, so 0300 drops and recreates rather than moving data it does not have. The Kafka topic and consumer group move with the tables: the producer that fills them is in this tree too, so both sides of the wire change in one commit, and the broker has no such topic to strand. tophog was already topfn everywhere the Python looks, and still tophog in the HCL layer, the schema snapshots and the skill that documents it -- a rename that stopped halfway leaves two names for one table, which is the thing it was meant to fix. The keeper paths now agree with what the engine derives from the table name. Both are refused in bin/debrand, so neither can walk back, and both are rules rather than edits, so a fresh upstream import lands on the new names. The one place hog_invocation_results is still sayable is the migration that retires it; that file is allowlisted by name. Co-authored-by: Hanzo Dev <[email protected]>
63 lines
2.7 KiB
TypeScript
63 lines
2.7 KiB
TypeScript
import { DateTime } from 'luxon'
|
|
|
|
import { KAFKA_FN_INVOCATION_RESULTS } from '~/common/config/kafka-topics'
|
|
import { KafkaProducerWrapper } from '~/common/kafka/producer'
|
|
import { Datastore } from '~/tests/helpers/datastore'
|
|
import { waitForExpect } from '~/tests/helpers/expectations'
|
|
|
|
// `~/common/kafka/producer` is auto-mocked in these tests; use the real wrapper to push probe rows.
|
|
const ActualKafkaProducerWrapper: typeof KafkaProducerWrapper =
|
|
jest.requireActual('~/common/kafka/producer').KafkaProducerWrapper
|
|
|
|
/**
|
|
* Probe invocations until a row lands, proving the Datastore Kafka engine's consumer has
|
|
* attached. Its `auto.offset.reset=latest` drops rows produced before attach, so a seed right after
|
|
* topic (re)creation can lose rows and never satisfy a count poll. Mirrors waitForDatastoreKafkaConsumer.
|
|
*/
|
|
export const waitForScriptInvocationResultsMvReady = async (datastore: Datastore): Promise<void> => {
|
|
const producer = await ActualKafkaProducerWrapper.create(undefined)
|
|
const probeTeamId = -999_999
|
|
try {
|
|
await waitForExpect(async () => {
|
|
await producer.queueMessages({
|
|
topic: KAFKA_FN_INVOCATION_RESULTS,
|
|
messages: [
|
|
{
|
|
key: 'probe',
|
|
value: JSON.stringify({
|
|
team_id: probeTeamId,
|
|
function_kind: 'insights_function',
|
|
function_id: 'probe',
|
|
invocation_id: 'probe',
|
|
parent_run_id: '',
|
|
status: 'running',
|
|
attempts: 0,
|
|
is_retry: 0,
|
|
scheduled_at: DateTime.utc().toFormat("yyyy-MM-dd HH:mm:ss.SSS'000'"),
|
|
started_at: null,
|
|
finished_at: null,
|
|
duration_ms: null,
|
|
error_kind: '',
|
|
error_message: '',
|
|
event_uuid: '',
|
|
distinct_id: '',
|
|
person_id: '',
|
|
invocation_globals: '{}',
|
|
version: String(BigInt(Date.now()) * 1000n),
|
|
is_deleted: 0,
|
|
}),
|
|
},
|
|
],
|
|
})
|
|
await producer.flush()
|
|
|
|
const result = await datastore.query<{ c: number }>(
|
|
`SELECT count() AS c FROM invocations WHERE team_id = ${probeTeamId}`
|
|
)
|
|
expect(Number(result[0]?.c ?? 0)).toBeGreaterThan(0)
|
|
}, 30_000)
|
|
} finally {
|
|
await producer.disconnect()
|
|
}
|
|
}
|