diff --git a/AGENTS.md b/AGENTS.md index 47919cfeda..71c59b479d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -165,6 +165,15 @@ $NAV --root backend callees "X" # what does X call? - Search for existing code to reuse before writing new code - Follow established patterns in the codebase - Keep changes focused — don't refactor beyond what's asked +- **A simpler design found late is still the design.** Work already spent is not an argument + for a shape, and neither is a clean review round, a passing suite, or a long PR thread. The + signal to stop and re-derive rather than patch again is a change that keeps growing to defend + its own structure: each review finding fixing an assumption the previous fix broke, the same + class of bug reappearing somewhere new, or most of the diff being consequences of one early + choice rather than the thing you set out to do. When that happens, say plainly what the + simpler design is and what switching costs — a migration, a review cycle restarted from zero, + work discarded — and let the user decide. Do not keep paying down the harder one because it + is nearly finished, and do not present the accumulated cost as a reason to continue. - **Ship only the tests the PR needs.** A committed test must pin behavior a future change could plausibly break, and be the smallest setup that exercises the new logic. While developing, write as many exhaustive tests and do as much manual testing as you need to convince yourself the change works — then remove that scaffolding before marking the PR ready, keeping only the essential regression guard(s). A test that merely re-exercises pre-existing behavior, or needs elaborate fixtures to assert something trivial, is scaffolding: delete it. If nothing meaningful is left to guard, ship no test rather than a ceremonial one. - **Comments record constraints, not narration.** Write a comment only for what the code can't show: why a non-obvious approach is required, what breaks if it's "simplified" away. State each invariant once, at the place where someone would break it, in ≤4 lines. Don't describe what the next line does, don't repeat the same rationale at multiple sites, and don't address the PR reviewer (justifying a change belongs in the PR description, not the code). Reference nothing ephemeral — no numbered steps from your dev flow, no "the poller / the test does X" scaffolding, no transient state that won't exist for the next reader; keep only the essential, durable rationale. Describe the code as it is, never its drafting history: "we no longer do X", "unchanged behavior", "instead of the previous approach" are meaningless to a reader who never saw the earlier iteration — before finishing, reread your comments as if the current state is the only state that ever existed. - **Never attribute work to a specific customer, account, or "requested by a customer" in repo-tracked content** (PR descriptions, commit messages, code comments, docs). Describe changes by their technical motivation instead. diff --git a/backend/.sqlx/query-624a7dbc6cc951a199b0e70d86c463a0e7b5248c226ee92d95df94c3099cc400.json b/backend/.sqlx/query-624a7dbc6cc951a199b0e70d86c463a0e7b5248c226ee92d95df94c3099cc400.json new file mode 100644 index 0000000000..3713af1486 --- /dev/null +++ b/backend/.sqlx/query-624a7dbc6cc951a199b0e70d86c463a0e7b5248c226ee92d95df94c3099cc400.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE log_file SET indexed_at = now()\n FROM unnest($1::text[], $2::text[]) AS c(hostname, file_path)\n WHERE log_file.indexed_at IS NULL\n AND log_file.hostname = c.hostname\n AND log_file.file_path = c.file_path", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "TextArray", + "TextArray" + ] + }, + "nullable": [] + }, + "hash": "624a7dbc6cc951a199b0e70d86c463a0e7b5248c226ee92d95df94c3099cc400" +} diff --git a/backend/.sqlx/query-b5c839baab25c4dcdd503d380cf7a886242277cd50555f20b2e22e13942d2a3a.json b/backend/.sqlx/query-6bbcb27a3bb70302076c559c8394b14b842f595f68dd885248abaaeabd2d0bf1.json similarity index 86% rename from backend/.sqlx/query-b5c839baab25c4dcdd503d380cf7a886242277cd50555f20b2e22e13942d2a3a.json rename to backend/.sqlx/query-6bbcb27a3bb70302076c559c8394b14b842f595f68dd885248abaaeabd2d0bf1.json index ba9f3814e8..13d1c9b1e8 100644 --- a/backend/.sqlx/query-b5c839baab25c4dcdd503d380cf7a886242277cd50555f20b2e22e13942d2a3a.json +++ b/backend/.sqlx/query-6bbcb27a3bb70302076c559c8394b14b842f595f68dd885248abaaeabd2d0bf1.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT\n hostname,\n mode::text,\n worker_group,\n log_ts,\n file_path,\n ok_lines,\n err_lines,\n json_fmt\n FROM log_file\n WHERE log_ts > $1\n ORDER BY log_ts ASC LIMIT $2", + "query": "SELECT\n hostname,\n mode::text,\n worker_group,\n log_ts,\n file_path,\n ok_lines,\n err_lines,\n json_fmt\n FROM log_file\n WHERE indexed_at IS NULL\n ORDER BY log_ts ASC, hostname ASC LIMIT $1", "describe": { "columns": [ { @@ -46,7 +46,6 @@ ], "parameters": { "Left": [ - "Timestamp", "Int8" ] }, @@ -61,5 +60,5 @@ true ] }, - "hash": "b5c839baab25c4dcdd503d380cf7a886242277cd50555f20b2e22e13942d2a3a" + "hash": "6bbcb27a3bb70302076c559c8394b14b842f595f68dd885248abaaeabd2d0bf1" } diff --git a/backend/.sqlx/query-8d207cc9ed101ff116b617d25a94633c1531170ded1fdf09114718b941f5e1db.json b/backend/.sqlx/query-8d207cc9ed101ff116b617d25a94633c1531170ded1fdf09114718b941f5e1db.json deleted file mode 100644 index d0b96444ee..0000000000 --- a/backend/.sqlx/query-8d207cc9ed101ff116b617d25a94633c1531170ded1fdf09114718b941f5e1db.json +++ /dev/null @@ -1,65 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT\n hostname,\n mode::text,\n worker_group,\n log_ts,\n file_path,\n ok_lines,\n err_lines,\n json_fmt\n FROM log_file\n WHERE log_ts > NOW() - make_interval(secs => $1)\n ORDER BY log_ts ASC LIMIT $2", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "hostname", - "type_info": "Varchar" - }, - { - "ordinal": 1, - "name": "mode", - "type_info": "Text" - }, - { - "ordinal": 2, - "name": "worker_group", - "type_info": "Varchar" - }, - { - "ordinal": 3, - "name": "log_ts", - "type_info": "Timestamp" - }, - { - "ordinal": 4, - "name": "file_path", - "type_info": "Varchar" - }, - { - "ordinal": 5, - "name": "ok_lines", - "type_info": "Int8" - }, - { - "ordinal": 6, - "name": "err_lines", - "type_info": "Int8" - }, - { - "ordinal": 7, - "name": "json_fmt", - "type_info": "Bool" - } - ], - "parameters": { - "Left": [ - "Float8", - "Int8" - ] - }, - "nullable": [ - false, - null, - true, - false, - false, - true, - true, - true - ] - }, - "hash": "8d207cc9ed101ff116b617d25a94633c1531170ded1fdf09114718b941f5e1db" -} diff --git a/backend/.sqlx/query-8e0461855d05dc03919c8979d8acdc85389c0629847d8fabb2ad0aa043957b2f.json b/backend/.sqlx/query-8e0461855d05dc03919c8979d8acdc85389c0629847d8fabb2ad0aa043957b2f.json new file mode 100644 index 0000000000..96745f9485 --- /dev/null +++ b/backend/.sqlx/query-8e0461855d05dc03919c8979d8acdc85389c0629847d8fabb2ad0aa043957b2f.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "WITH moved AS (\n UPDATE log_file SET indexed_at = CASE\n WHEN log_ts > NOW() - make_interval(secs => $1) THEN NULL\n ELSE now() END\n WHERE indexed_at = 'epoch' RETURNING 1)\n SELECT count(*) AS \"n!\" FROM moved", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "n!", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Float8" + ] + }, + "nullable": [ + null + ] + }, + "hash": "8e0461855d05dc03919c8979d8acdc85389c0629847d8fabb2ad0aa043957b2f" +} diff --git a/backend/.sqlx/query-947f7ca06f6f9a3fd50f817bc9b0c06924719189bec8e5c19866db8d0b87df5e.json b/backend/.sqlx/query-947f7ca06f6f9a3fd50f817bc9b0c06924719189bec8e5c19866db8d0b87df5e.json new file mode 100644 index 0000000000..6739d80033 --- /dev/null +++ b/backend/.sqlx/query-947f7ca06f6f9a3fd50f817bc9b0c06924719189bec8e5c19866db8d0b87df5e.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE log_file SET indexed_at = now()\n FROM unnest($1::text[], $2::timestamp[]) AS c(hostname, log_ts)\n WHERE log_file.hostname = c.hostname AND log_file.log_ts = c.log_ts", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "TextArray", + "TimestampArray" + ] + }, + "nullable": [] + }, + "hash": "947f7ca06f6f9a3fd50f817bc9b0c06924719189bec8e5c19866db8d0b87df5e" +} diff --git a/backend/.sqlx/query-a7d450e34084d561f69e588bd76fd56e616ee79d895b7dcc37ad9442789e1574.json b/backend/.sqlx/query-a7d450e34084d561f69e588bd76fd56e616ee79d895b7dcc37ad9442789e1574.json new file mode 100644 index 0000000000..a83c34e0db --- /dev/null +++ b/backend/.sqlx/query-a7d450e34084d561f69e588bd76fd56e616ee79d895b7dcc37ad9442789e1574.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "WITH retired AS (\n UPDATE log_file SET indexed_at = now()\n WHERE indexed_at IS NULL\n AND log_ts <= NOW() - make_interval(secs => $1) RETURNING 1)\n SELECT count(*) AS \"n!\" FROM retired", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "n!", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Float8" + ] + }, + "nullable": [ + null + ] + }, + "hash": "a7d450e34084d561f69e588bd76fd56e616ee79d895b7dcc37ad9442789e1574" +} diff --git a/backend/.sqlx/query-b8e732a03969666444f73397ac153d23ec2af084b2f93da24c920532c1916384.json b/backend/.sqlx/query-b8e732a03969666444f73397ac153d23ec2af084b2f93da24c920532c1916384.json index b336210daf..b892061f56 100644 --- a/backend/.sqlx/query-b8e732a03969666444f73397ac153d23ec2af084b2f93da24c920532c1916384.json +++ b/backend/.sqlx/query-b8e732a03969666444f73397ac153d23ec2af084b2f93da24c920532c1916384.json @@ -98,12 +98,12 @@ null, null, null, - true, + false, null, null, null, - true, - true + false, + false ] }, "hash": "b8e732a03969666444f73397ac153d23ec2af084b2f93da24c920532c1916384" diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 8e49c42d98..693921efc6 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -58738c39ac41d57917bbd9400318704763d997f7 +bc0c7051585194474078b6c1941a3fb73893d9e5 diff --git a/backend/migrations/20260830085453_log_file_indexed_at.down.sql b/backend/migrations/20260830085453_log_file_indexed_at.down.sql new file mode 100644 index 0000000000..ac3c92f3eb --- /dev/null +++ b/backend/migrations/20260830085453_log_file_indexed_at.down.sql @@ -0,0 +1,4 @@ +DROP INDEX IF EXISTS index_log_file_premigration; +DROP INDEX IF EXISTS index_log_file_pending_path; +DROP INDEX IF EXISTS index_log_file_pending; +ALTER TABLE log_file DROP COLUMN IF EXISTS indexed_at; diff --git a/backend/migrations/20260830085453_log_file_indexed_at.up.sql b/backend/migrations/20260830085453_log_file_indexed_at.up.sql new file mode 100644 index 0000000000..3421a5082a --- /dev/null +++ b/backend/migrations/20260830085453_log_file_indexed_at.up.sql @@ -0,0 +1,35 @@ +-- The service log ingest walked `log_file` with a cursor over `log_ts`, which is when a line +-- was written rather than when its row appeared. Rows do not arrive in that order — an upload +-- retried after a failure, a host that has just started, a batch the row limit cut mid-minute — +-- and a row that becomes visible behind the cursor is never read: it stays in `log_file` and its +-- lines stay out of search until retention drops them. +-- +-- No ordering fixes this. A cursor over arrival order fails the same way, because `nextval` is +-- allocated before its INSERT commits: a row can be assigned a lower value and commit after a +-- higher one has already moved the cursor past it. Which rows are outstanding is a property of +-- the rows, so it is recorded on them. +ALTER TABLE log_file ADD COLUMN indexed_at TIMESTAMPTZ; + +-- Rows that already existed are marked, not queued: on a 14-day window most were ingested long +-- ago and their raw files are gone. A sentinel rather than a timestamp, because the indexer has to +-- tell them apart from rows registered since — those start NULL — and it puts the window's worth of +-- them back on the queue on its first pass, keeping only what the columnar store can vouch for. +-- +-- Not split here on the cursor the old ingest had reached. Below that cursor sits every row it +-- skipped, which is the loss this migration exists to stop; recording those as done would carry the +-- bug into its own fix. +UPDATE log_file SET indexed_at = 'epoch' WHERE indexed_at IS NULL; + +-- The work queue, and the only index the ingest query needs: outstanding rows are a small +-- fraction of the table, so this stays proportional to what is left to do rather than to the +-- retention window. +CREATE INDEX index_log_file_pending ON log_file (log_ts) WHERE indexed_at IS NULL; + +-- A rebuild takes rows out of the queue by the file it just read out of the store, which is +-- the one lookup that arrives without a `log_ts`: the primary key is `(hostname, log_ts)`, so +-- nothing else covers it and each batch would scan every outstanding row instead. +CREATE INDEX index_log_file_pending_path ON log_file (hostname, file_path) WHERE indexed_at IS NULL; + +-- Reached once per pass while pre-migration rows survive, and never again after the first +-- conversion clears them. +CREATE INDEX index_log_file_premigration ON log_file (log_ts) WHERE indexed_at = 'epoch'; diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index c0748a3798..ce09cebe71 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -1358,9 +1358,9 @@ fn last_log_file_sent() -> Option { /// the file that was still open and the appender reopens that minute in append mode, /// so a restart inside it would otherwise strand everything written afterwards. /// -/// A row rewritten this way restores the object and sums the counters, but whether the -/// indexers read it again depends on their single `log_ts >` cursor, which is not -/// per-hostname: a minute at or below it stays out of search until it is re-indexed. +/// A row rewritten this way restores the object and sums the counters, but it keeps the +/// `indexed_at` it already had, so one the indexers have taken is not offered again and +/// the lines added by the rewrite stay out of search. async fn init_last_log_file_sent(conn: &Connection, hostname: &str) { let Some(db) = conn.as_sql() else { return; @@ -1406,9 +1406,10 @@ async fn send_log_files_to_object_store( if ts < retention_cutoff { continue; } - // Stop at the first failure rather than moving on: both indexers walk - // `log_file` with a `log_ts > watermark` cursor, so a row that lands after - // a newer one is never picked up. + // Stop at the first failure rather than moving on, so a file is never + // registered before an older one that has not made it to the store yet. + // The indexers do not depend on that ordering — every row is offered until + // it is marked — but a gap here would still be visible while it lasts. if !send_log_file_to_object_store(hostname, mode, worker_group, conn, &file_name, ts).await { break; diff --git a/backend/summarized_schema.txt b/backend/summarized_schema.txt index a9865cebd3..607ef99e0c 100644 --- a/backend/summarized_schema.txt +++ b/backend/summarized_schema.txt @@ -128,7 +128,7 @@ job_stats: workspace_id(char), job_id(uuid), metric_id(char), metric_name(char), kafka_pending_commits: id(bigint), workspace_id(char), kafka_trigger_path(char), topic(char), partition(int), offset(bigint), created_at(ts) FK: (workspace_id, kafka_trigger_path) -> kafka_trigger(workspace_id, path) kafka_trigger: path(char), kafka_resource_path(char), topics(char), group_id(char), script_path(char), is_flow(bool), workspace_id(char), edited_by(char), email(char), edited_at(ts), extra_perms(jsonb), server_id(char), last_server_ping(ts), error(text), error_handler_path(char), error_handler_args(jsonb), retry(jsonb), mode(trigger_mode), filters(jsonb[]), auto_commit(bool), labels(text[]) -log_file: hostname(char), log_ts(ts), ok_lines(bigint), err_lines(bigint), mode(log_mode), worker_group(char), file_path(char), json_fmt(bool) +log_file: hostname(char), log_ts(ts), ok_lines(bigint), err_lines(bigint), mode(log_mode), worker_group(char), file_path(char), json_fmt(bool), indexed_at(ts) macro_definition: workspace_id(char), name(char), provider_path(char), params(text), body(text), is_table_macro(bool), created_at(ts) FK: (workspace_id) -> workspace(id) macro_usage: workspace_id(char), consumer_path(char), macro_name(char)