fix: harden the service log indexer's recovery and read paths (#10904)

* [ee] fix: an unreadable ingest cursor should not stop the server booting

Three follow-ups to #10894, all in the service log indexer: a corrupt cursor no
longer takes the server down at boot, the queue's writes are covered against a
real database rather than by hand, and a read skips the dedupe when the partition
it reads holds a single object.

* [ee] test: place the queue's rows relative to the clock the statement reads

Also drops the two `.sqlx` entries the query extraction orphaned: sqlx keys on the
literal including its indentation, so moving a query into a function leaves the
old copy behind.

* [ee] test: make the pair-exactness and rebuild-dedupe tests actually bite

* [ee] docs: state the cursor and dedupe rules without their history

* chore: update ee-repo-ref to 90a368362896ebcc2fcfaaf9510dc9be68c929f7

This commit updates the EE repository reference after PR #761 was merged in windmill-ee-private.

Previous ee-repo-ref: e3423705aa8f2d585bc65474cfd0c4c762ec4ad5

New ee-repo-ref: 90a368362896ebcc2fcfaaf9510dc9be68c929f7

Automated by sync-ee-ref workflow.

---------

Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>
This commit is contained in:
Ruben Fiszel
2026-08-31 16:01:28 +00:00
committed by GitHub
co-authored by windmill-internal-app[bot]
parent b57e231c2b
commit 831370cdde
5 changed files with 38 additions and 38 deletions
@@ -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": "872be44352d4a27a2005f6bdc38f927ac4886755fd17ec755037763aa92f6c0e"
}
@@ -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": "87e8e034b1cf5ea7ce43870d77d33fa0fd05b067454f0d4686d8b155fe562ffe"
}
@@ -1,22 +0,0 @@
{
"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"
}
@@ -1,15 +0,0 @@
{
"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"
}
+1 -1
View File
@@ -1 +1 @@
bc0c7051585194474078b6c1941a3fb73893d9e5
90a368362896ebcc2fcfaaf9510dc9be68c929f7