Merge remote-tracking branch 'origin/main' into http-trigger-cors-config

2026-09-21 00:02:30 +00:00 · 2026-08-31 15:19:49 +02:00
parent 89ca65e3aa aa4a6ffd66
commit 74ca65fc23
35 changed files with 1034 additions and 411 deletions
@@ -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.
@@ -1,6 +1,6 @@
 {
  "db_name": "PostgreSQL",
-  "query": "DELETE FROM log_file WHERE log_ts <= now() - ($1::bigint::text || ' s')::interval RETURNING file_path, hostname",
+  "query": "DELETE FROM log_file WHERE (hostname, log_ts) IN (\n                 SELECT hostname, log_ts FROM log_file\n                 WHERE log_ts <= now() - ($1::bigint::text || ' s')::interval\n                 LIMIT $2\n             ) RETURNING file_path, hostname",
  "describe": {
    "columns": [
      {
@@ -16,6 +16,7 @@
    ],
    "parameters": {
      "Left": [
+        "Int8",
        "Int8"
      ]
    },
@@ -24,5 +25,5 @@
      false
    ]
  },
-  "hash": "94da1e7feb4f58cc7ebe99752736f956d47810a94cb052fdcffb5cfe440f8033"
+  "hash": "0e03dc960c0a22e042e54af719ac90c4b8506acefc85ef5b2f92a7bc451b1c5e"
 }
@@ -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"
+}
@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT max(log_ts) FROM log_file\n         WHERE hostname = $1 AND log_ts < (SELECT max(log_ts) FROM log_file WHERE hostname = $1)",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "max",
+        "type_info": "Timestamp"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "67a83afb708c90b2132cba81a0701dfc8b5e7aedf2f26fc1fe5be8f685cf709c"
+}
@@ -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"
 }
@@ -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"
-}
@@ -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"
+}
@@ -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"
+}
@@ -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"
+}
@@ -98,12 +98,12 @@
      null,
      null,
      null,
-      true,
+      false,
      null,
      null,
      null,
-      true,
-      true
+      false,
+      false
    ]
  },
  "hash": "b8e732a03969666444f73397ac153d23ec2af084b2f93da24c920532c1916384"
@@ -0,0 +1,29 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT mode::text AS \"mode!\", log_ts FROM log_file WHERE hostname = $1 AND file_path = $2 ORDER BY log_ts DESC LIMIT 1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "mode!",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 1,
+        "name": "log_ts",
+        "type_info": "Timestamp"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null,
+      false
+    ]
+  },
+  "hash": "daa5b57290cd1f821a53eebe96434f1befe6b16eee363c843faa1f836d53ca8d"
+}
@@ -1,6 +1,6 @@
 {
  "db_name": "PostgreSQL",
-  "query": "INSERT INTO log_file (hostname, mode, worker_group, log_ts, file_path, ok_lines, err_lines, json_fmt)\n             VALUES ($1, $2::text::LOG_MODE, $3, $4, $5, $6, $7, $8)\n             ON CONFLICT (hostname, log_ts) DO UPDATE SET ok_lines = log_file.ok_lines + $6, err_lines = log_file.err_lines + $7",
+  "query": "INSERT INTO log_file (hostname, mode, worker_group, log_ts, file_path, ok_lines, err_lines, json_fmt)\n     VALUES ($1, $2::text::LOG_MODE, $3, $4, $5, $6, $7, $8)\n     ON CONFLICT (hostname, log_ts) DO UPDATE SET ok_lines = log_file.ok_lines + $6, err_lines = log_file.err_lines + $7",
  "describe": {
    "columns": [],
    "parameters": {
@@ -17,5 +17,5 @@
    },
    "nullable": []
  },
-  "hash": "92faee8714a45a403b623e04d789f7f99067a05e9dfe270223164db8a1df2e4b"
+  "hash": "f277db0459ff311d8a396aa4e03876dac75ba6df5f8eaa9f185300483e3ee36f"
 }
@@ -3731,9 +3731,9 @@ dependencies = [

 [[package]]
 name = "datasketches"
-version = "0.2.0"
+version = "0.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c286de4e81ea2590afc24d754e0f83810c566f50a1388fa75ebd57928c0d9745"
+checksum = "46c4cf71a36b46dcfc00e5014c0c20ccad2b1b6a008304d7d57d2749b2d41b3d"

 [[package]]
 name = "debug-helper"
@@ -5175,6 +5175,12 @@ dependencies = [
 "syn 2.0.119",
 ]

+[[package]]
+name = "frostem"
+version = "1.20260821.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "08ed6437a2ed7fc408115e25cdb41e15ba4b17742a4c8d3d1b5276fe9b687209"
+
 [[package]]
 name = "fs3"
 version = "0.5.0"
@@ -7379,15 +7385,6 @@ dependencies = [
 "hashbrown 0.15.5",
 ]

-[[package]]
-name = "lru"
-version = "0.16.4"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7f66e8d5d03f609abc3a39e6f08e4164ebf1447a732906d39eb9b99b7919ef39"
-dependencies = [
- "hashbrown 0.16.1",
-]
-
 [[package]]
 name = "lru"
 version = "0.18.3"
@@ -7423,9 +7420,9 @@ dependencies = [

 [[package]]
 name = "lz4_flex"
-version = "0.13.1"
+version = "0.14.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7ef0d4ed8669f8f8826eb00dc878084aa8f253506c4fd5e8f58f5bce72ddb97e"
+checksum = "ecbdfe44b1bd960b68170b417450a628c43f7cf56bb3c5317e61cb230ee7f226"

 [[package]]
 name = "lzma-sys"
@@ -7849,9 +7846,9 @@ dependencies = [

 [[package]]
 name = "murmurhash32"
-version = "0.3.1"
+version = "0.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2195bf6aa996a481483b29d62a7663eed3fe39600c460e323f8ff41e90bdd89b"
+checksum = "a8afc6df942f4c022d70c5725e18df1a705773870e5b7f96fde94ca0334ce77a"

 [[package]]
 name = "mysql-common-derive"
@@ -8842,7 +8839,7 @@ checksum = "1a80800c0488c3a21695ea981a54918fbb37abf04f4d0720c453632255e2ff0e"
 [[package]]
 name = "ownedbytes"
 version = "0.9.0"
-source = "git+https://github.com/windmill-labs/tantivy?rev=6ae7c70bc603b8e69e27f3240e08bd00a93fb12c#6ae7c70bc603b8e69e27f3240e08bd00a93fb12c"
+source = "git+https://github.com/windmill-labs/tantivy?rev=ea3b818c7b93db0c2b5db4f5e7ffef38f339060a#ea3b818c7b93db0c2b5db4f5e7ffef38f339060a"
 dependencies = [
 "stable_deref_trait",
 ]
@@ -10660,16 +10657,6 @@ dependencies = [
 "walkdir",
 ]

-[[package]]
-name = "rust-stemmers"
-version = "1.2.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e46a2036019fdb888131db7a4c847a1063a7493f971ed94ea82c67eada63ca54"
-dependencies = [
- "serde",
- "serde_derive",
-]
-
 [[package]]
 name = "rust_decimal"
 version = "1.42.1"
@@ -12803,12 +12790,12 @@ checksum = "7b2093cf4c8eb1e67749a6762251bc9cd836b6fc171623bd0a9d324d37af2417"

 [[package]]
 name = "tantivy"
-version = "0.26.0"
-source = "git+https://github.com/windmill-labs/tantivy?rev=6ae7c70bc603b8e69e27f3240e08bd00a93fb12c#6ae7c70bc603b8e69e27f3240e08bd00a93fb12c"
+version = "0.27.0"
+source = "git+https://github.com/windmill-labs/tantivy?rev=ea3b818c7b93db0c2b5db4f5e7ffef38f339060a#ea3b818c7b93db0c2b5db4f5e7ffef38f339060a"
 dependencies = [
 "aho-corasick",
 "arc-swap",
- "base64 0.22.1",
+ "base64 0.23.1",
 "bitpacking",
 "bon",
 "byteorder",
@@ -12819,20 +12806,20 @@ dependencies = [
 "downcast-rs",
 "fastdivide",
 "fnv",
+ "frostem",
 "fs4",
 "htmlescape",
 "itertools 0.14.0",
 "levenshtein_automata",
 "log",
- "lru 0.16.4",
- "lz4_flex 0.13.1",
+ "lru 0.18.3",
+ "lz4_flex 0.14.0",
 "measure_time",
 "memmap2",
 "once_cell",
 "oneshot",
 "rayon",
 "regex",
- "rust-stemmers",
 "rustc-hash 2.1.3",
 "serde",
 "serde_json",
@@ -12849,22 +12836,23 @@ dependencies = [
 "thiserror 2.0.20",
 "time",
 "typetag",
+ "unwrap-infallible",
 "uuid",
 "winapi",
 ]

 [[package]]
 name = "tantivy-bitpacker"
-version = "0.9.0"
-source = "git+https://github.com/windmill-labs/tantivy?rev=6ae7c70bc603b8e69e27f3240e08bd00a93fb12c#6ae7c70bc603b8e69e27f3240e08bd00a93fb12c"
+version = "0.10.0"
+source = "git+https://github.com/windmill-labs/tantivy?rev=ea3b818c7b93db0c2b5db4f5e7ffef38f339060a#ea3b818c7b93db0c2b5db4f5e7ffef38f339060a"
 dependencies = [
 "bitpacking",
 ]

 [[package]]
 name = "tantivy-columnar"
-version = "0.6.0"
-source = "git+https://github.com/windmill-labs/tantivy?rev=6ae7c70bc603b8e69e27f3240e08bd00a93fb12c#6ae7c70bc603b8e69e27f3240e08bd00a93fb12c"
+version = "0.7.0"
+source = "git+https://github.com/windmill-labs/tantivy?rev=ea3b818c7b93db0c2b5db4f5e7ffef38f339060a#ea3b818c7b93db0c2b5db4f5e7ffef38f339060a"
 dependencies = [
 "downcast-rs",
 "fastdivide",
@@ -12878,8 +12866,8 @@ dependencies = [

 [[package]]
 name = "tantivy-common"
-version = "0.10.0"
-source = "git+https://github.com/windmill-labs/tantivy?rev=6ae7c70bc603b8e69e27f3240e08bd00a93fb12c#6ae7c70bc603b8e69e27f3240e08bd00a93fb12c"
+version = "0.11.0"
+source = "git+https://github.com/windmill-labs/tantivy?rev=ea3b818c7b93db0c2b5db4f5e7ffef38f339060a#ea3b818c7b93db0c2b5db4f5e7ffef38f339060a"
 dependencies = [
 "async-trait",
 "byteorder",
@@ -12901,8 +12889,8 @@ dependencies = [

 [[package]]
 name = "tantivy-query-grammar"
-version = "0.25.0"
-source = "git+https://github.com/windmill-labs/tantivy?rev=6ae7c70bc603b8e69e27f3240e08bd00a93fb12c#6ae7c70bc603b8e69e27f3240e08bd00a93fb12c"
+version = "0.26.0"
+source = "git+https://github.com/windmill-labs/tantivy?rev=ea3b818c7b93db0c2b5db4f5e7ffef38f339060a#ea3b818c7b93db0c2b5db4f5e7ffef38f339060a"
 dependencies = [
 "fnv",
 "nom",
@@ -12913,8 +12901,8 @@ dependencies = [

 [[package]]
 name = "tantivy-sstable"
-version = "0.6.0"
-source = "git+https://github.com/windmill-labs/tantivy?rev=6ae7c70bc603b8e69e27f3240e08bd00a93fb12c#6ae7c70bc603b8e69e27f3240e08bd00a93fb12c"
+version = "0.7.0"
+source = "git+https://github.com/windmill-labs/tantivy?rev=ea3b818c7b93db0c2b5db4f5e7ffef38f339060a#ea3b818c7b93db0c2b5db4f5e7ffef38f339060a"
 dependencies = [
 "futures-util",
 "itertools 0.14.0",
@@ -12926,8 +12914,8 @@ dependencies = [

 [[package]]
 name = "tantivy-stacker"
-version = "0.6.0"
-source = "git+https://github.com/windmill-labs/tantivy?rev=6ae7c70bc603b8e69e27f3240e08bd00a93fb12c#6ae7c70bc603b8e69e27f3240e08bd00a93fb12c"
+version = "0.7.0"
+source = "git+https://github.com/windmill-labs/tantivy?rev=ea3b818c7b93db0c2b5db4f5e7ffef38f339060a#ea3b818c7b93db0c2b5db4f5e7ffef38f339060a"
 dependencies = [
 "murmurhash32",
 "tantivy-common",
@@ -12935,8 +12923,8 @@ dependencies = [

 [[package]]
 name = "tantivy-tokenizer-api"
-version = "0.6.0"
-source = "git+https://github.com/windmill-labs/tantivy?rev=6ae7c70bc603b8e69e27f3240e08bd00a93fb12c#6ae7c70bc603b8e69e27f3240e08bd00a93fb12c"
+version = "0.7.0"
+source = "git+https://github.com/windmill-labs/tantivy?rev=ea3b818c7b93db0c2b5db4f5e7ffef38f339060a#ea3b818c7b93db0c2b5db4f5e7ffef38f339060a"
 dependencies = [
 "serde",
 ]
@@ -14257,6 +14245,12 @@ version = "0.9.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1"

+[[package]]
+name = "unwrap-infallible"
+version = "1.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e497bb1f828cc9fb236722c2eaa100dcf201563f38f4da6252357a59037adf31"
+
 [[package]]
 name = "ureq"
 version = "2.12.1"
@@ -15693,9 +15687,11 @@ dependencies = [
 "bytes",
 "chrono",
 "const_format",
+ "datafusion",
 "flume",
 "futures",
 "lazy_static",
+ "object_store",
 "serde",
 "serde_json",
 "sqlx",
@@ -15703,6 +15699,7 @@ dependencies = [
 "tempfile",
 "tokio",
 "tracing",
+ "url",
 "uuid",
 "windmill-common",
 "windmill-object-store",
@@ -477,7 +477,10 @@ rust-embed = { version = "^6", features = ["interpolate-folder-path"] }
 mime_guess = "^2"
 hex = "^0"
 sql-builder = "^3"
-argon2 = "^0"
+# Minor-pinned rather than the `^0` used elsewhere in this file: argon2's 0.x
+# minors are API-breaking (0.6 moved `SaltString` into `phc`, put `rand_core`
+# behind a feature and changed `hash_password`), so a float breaks the build.
+argon2 = "0.6"
 quick_cache = "^0"
 rand = "=0.9.0"
 rand_core = { version = "^0", features = ["std"] }
@@ -697,7 +700,7 @@ tikv-jemalloc-ctl = { version = "^0.5" }
 triomphe = "^0"
 pin-project-lite = "^0"

-tantivy = { git="https://github.com/windmill-labs/tantivy", rev="6ae7c70bc603b8e69e27f3240e08bd00a93fb12c" }
+tantivy = { git="https://github.com/windmill-labs/tantivy", rev="ea3b818c7b93db0c2b5db4f5e7ffef38f339060a" }

 backon = "1.3.0"

@@ -1 +1 @@
-9ff97cd818e85940fec282c92161e98c1b8583e2
+bc0c7051585194474078b6c1941a3fb73893d9e5
@@ -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;
@@ -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';
@@ -14,7 +14,7 @@ use monitor::{
    reload_nuget_config_setting, reload_powershell_repo_pat_setting,
    reload_powershell_repo_url_setting, reload_ruby_repos_setting,
    reload_timeout_wait_result_setting, reload_workspace_registries_setting,
-    send_current_log_file_to_object_store, send_logs_to_object_store, WORKERS_NAMES,
+    flush_pending_log_files_to_object_store, send_logs_to_object_store, WORKERS_NAMES,
 };
 use rand::Rng;
 use sqlx::{Pool, Postgres};
@@ -62,8 +62,9 @@ use windmill_common::{
        SAML_METADATA_SETTING, SANDBOX_IMAGE_CACHE_MAX_MB_SETTING,
        SANDBOX_IMAGE_DEFAULT_REGISTRY_SETTING, SANDBOX_IMAGE_MAX_SIZE_MB_SETTING,
        SANDBOX_IMAGE_PULL_POLICY_SETTING, SANDBOX_REGISTRY_AUTH_SETTING, SCIM_TOKEN_SETTING,
-        SMTP_SETTING, STORE_AUDIT_LOGS_S3_SETTING, TEAMS_SETTING, TIMEOUT_WAIT_RESULT_SETTING,
-        UV_EXCLUDE_NEWER_SETTING, UV_INDEX_STRATEGY_SETTING, UV_PYTHON_INSTALL_MIRROR_SETTING,
+        SERVICE_LOG_RETENTION_SECS_SETTING, SMTP_SETTING, STORE_AUDIT_LOGS_S3_SETTING,
+        TEAMS_SETTING, TIMEOUT_WAIT_RESULT_SETTING, UV_EXCLUDE_NEWER_SETTING,
+        UV_INDEX_STRATEGY_SETTING, UV_PYTHON_INSTALL_MIRROR_SETTING,
        WORKSPACE_FAIRNESS_DURATION_SECS_SETTING, WORKSPACE_FAIRNESS_ENABLED_SETTING,
        WORKSPACE_FAIRNESS_MAX_PERCENT_SETTING, WORKSPACE_FAIRNESS_MIN_TOTAL_SETTING,
        WORKSPACE_MAX_QUEUED_JOBS_SETTING, WORKSPACE_REGISTRIES_SETTING,
@@ -145,7 +146,8 @@ use crate::monitor::{
    reload_pip_index_url_setting, reload_retention_period_setting,
    reload_sandbox_image_cache_max_setting, reload_sandbox_image_default_registry_setting,
    reload_sandbox_image_max_size_setting, reload_sandbox_image_pull_policy_setting,
-    reload_sandbox_registry_auth_setting, reload_scim_token_setting, reload_smtp_config,
+    reload_sandbox_registry_auth_setting, reload_scim_token_setting,
+    reload_service_log_retention_secs_setting, reload_smtp_config,
    reload_store_audit_logs_s3_setting, reload_uv_exclude_newer_setting,
    reload_uv_index_strategy_setting, reload_uv_python_install_mirror_setting,
    reload_worker_config, MonitorIteration,
@@ -1264,10 +1266,12 @@ Windmill Community Edition {GIT_VERSION}
        #[cfg(all(feature = "tantivy", feature = "parquet"))]
        let log_indexer_f = {
            let log_indexer_rx = killpill_rx.resubscribe();
-            let log_index_writer2 = log_index_writer.clone();
+            // Moved, not cloned: sealing a chunk takes sole ownership of its
+            // tantivy writer, which a second live handle would silently prevent.
+            let moved_log_index_writer = log_index_writer;
            async {
                if let Some(db) = conn.as_sql() {
-                    if let Some(log_index_writer) = log_index_writer2 {
+                    if let Some(log_index_writer) = moved_log_index_writer {
                        windmill_indexer::service_logs_oss::run_indexer(
                            db.clone(),
                            log_index_writer,
@@ -1664,7 +1668,7 @@ Windmill Community Edition {GIT_VERSION}
    } else {
        tracing::info!("Nothing to do, exiting.");
    }
-    send_current_log_file_to_object_store(&conn, &hostname, &mode).await;
+    flush_pending_log_files_to_object_store(&conn, &hostname, &mode).await;

    if let Some(db) = conn.as_sql() {
        tracing::info!("Exiting connection pool");
@@ -1955,6 +1959,9 @@ async fn process_notify_event(
                }
                TIMEOUT_WAIT_RESULT_SETTING => reload_timeout_wait_result_setting(conn).await,
                RETENTION_PERIOD_SECS_SETTING => reload_retention_period_setting(conn).await,
+                SERVICE_LOG_RETENTION_SECS_SETTING => {
+                    reload_service_log_retention_secs_setting(conn).await
+                }
                RETENTION_PERIOD_SECS_OVERRIDES_SETTING => {
                    if let Err(e) = load_retention_period_overrides(db).await {
                        tracing::error!("Error loading per-workspace retention overrides: {e:#}");
@@ -70,18 +70,17 @@ use windmill_common::{
        RETENTION_PERIOD_SECS_SETTING, SAML_METADATA_SETTING, SANDBOX_IMAGE_CACHE_MAX_MB_SETTING,
        SANDBOX_IMAGE_DEFAULT_REGISTRY_SETTING, SANDBOX_IMAGE_MAX_SIZE_MB_SETTING,
        SANDBOX_IMAGE_PULL_POLICY_SETTING, SANDBOX_REGISTRY_AUTH_SETTING, SCIM_TOKEN_SETTING,
-        SMTP_SETTING, STORE_AUDIT_LOGS_S3_SETTING, TIMEOUT_WAIT_RESULT_SETTING,
-        UV_EXCLUDE_NEWER_SETTING, UV_INDEX_STRATEGY_SETTING, UV_PYTHON_INSTALL_MIRROR_SETTING,
-        WORKSPACE_FAIRNESS_DURATION_SECS_SETTING, WORKSPACE_FAIRNESS_ENABLED_SETTING,
-        WORKSPACE_FAIRNESS_MAX_PERCENT_SETTING, WORKSPACE_FAIRNESS_MIN_TOTAL_SETTING,
-        WORKSPACE_MAX_QUEUED_JOBS_SETTING,
+        SERVICE_LOG_RETENTION_SECS_SETTING, SMTP_SETTING, STORE_AUDIT_LOGS_S3_SETTING,
+        TIMEOUT_WAIT_RESULT_SETTING, UV_EXCLUDE_NEWER_SETTING, UV_INDEX_STRATEGY_SETTING,
+        UV_PYTHON_INSTALL_MIRROR_SETTING, WORKSPACE_FAIRNESS_DURATION_SECS_SETTING,
+        WORKSPACE_FAIRNESS_ENABLED_SETTING, WORKSPACE_FAIRNESS_MAX_PERCENT_SETTING,
+        WORKSPACE_FAIRNESS_MIN_TOTAL_SETTING, WORKSPACE_MAX_QUEUED_JOBS_SETTING,
    },
    indexer::load_indexer_config,
    jobs::delete_jobs,
    jwt::JWT_SECRET,
    oauth2::REQUIRE_PREEXISTING_USER_FOR_OAUTH,
    server::load_smtp_config,
-    tracing_init::JSON_FMT,
    users::truncate_token,
    utils::{empty_as_none, now_from_db, report_critical_error, Mode, HUB_API_SECRET},
    worker::{
@@ -98,10 +97,10 @@ use windmill_common::{
    KillpillSender, AUDIT_LOG_RETENTION_DAYS, BASE_URL, CRITICAL_ALERTS_ON_DB_OVERSIZE,
    CRITICAL_ALERTS_ON_TOKEN_EXPIRY, CRITICAL_ALERT_MUTE_UI_ENABLED,
    CRITICAL_ALERT_MUTE_ZOMBIE_JOB_RESTART, CRITICAL_ERROR_CHANNELS, DB, DEFAULT_HUB_BASE_URL,
-    HUB_BASE_URL, JOB_RETENTION_SECS, JOB_RETENTION_SECS_OVERRIDES,
-    JOB_RETENTION_SECS_OVERRIDES_LOADED, METRICS_DEBUG_ENABLED, METRICS_ENABLED,
-    MONITOR_LOGS_ON_OBJECT_STORE, OTEL_LOGS_ENABLED, OTEL_METRICS_ENABLED, OTEL_TRACING_ENABLED,
-    SERVICE_LOG_RETENTION_SECS, STORE_AUDIT_LOGS_S3,
+    DEFAULT_SERVICE_LOG_RETENTION_SECS, HUB_BASE_URL, JOB_RETENTION_SECS,
+    JOB_RETENTION_SECS_OVERRIDES, JOB_RETENTION_SECS_OVERRIDES_LOADED, METRICS_DEBUG_ENABLED,
+    METRICS_ENABLED, MONITOR_LOGS_ON_OBJECT_STORE, OTEL_LOGS_ENABLED, OTEL_METRICS_ENABLED,
+    OTEL_TRACING_ENABLED, STORE_AUDIT_LOGS_S3,
 };
 use windmill_common::{
    client::AuthedClient,
@@ -489,6 +488,19 @@ pub async fn initial_load(
        |v: Option<String>| async move { HUB_API_SECRET.store(std::sync::Arc::new(v)) },
    );

+    // Outside the `server_mode` guard below: every mode reads this. A worker registers its
+    // rotated log files against the cutoff, and a dedicated indexer trims the search index to a
+    // window derived from it — neither is a server.
+    pass.setting(SERVICE_LOG_RETENTION_SECS_SETTING, true, |v| async move {
+        windmill_common::set_service_log_retention_secs(parse_setting_value::<i64>(
+            v,
+            SERVICE_LOG_RETENTION_SECS_SETTING,
+            "SERVICE_LOG_RETENTION_SECS",
+            DEFAULT_SERVICE_LOG_RETENTION_SECS,
+            |x| x,
+        ))
+    });
+
    if server_mode {
        pass.setting(RETENTION_PERIOD_SECS_SETTING, true, |v| async move {
            JOB_RETENTION_SECS.store(
@@ -1234,32 +1246,60 @@ async fn sleep_until_next_minute_start_plus_one_s() {
 }

 use windmill_common::tracing_init::TMP_WINDMILL_LOGS_SERVICE;
-async fn find_two_highest_files(hostname: &str) -> (Option<String>, Option<String>) {
+
+/// The minutely rolling appender names each file `<hostname>.log.<%Y-%m-%d-%H-%M>`;
+/// anything else in the directory is not a rotated log file.
+fn parse_log_file_ts(file_name: &str) -> Option<NaiveDateTime> {
+    NaiveDateTime::parse_from_str(
+        file_name.rsplit('.').next()?,
+        windmill_common::tracing_init::LOG_TIMESTAMP_FMT,
+    )
+    .ok()
+}
+
+/// Oldest first. Readdir order is filesystem-dependent — tmpfs hands back the
+/// newest entry first, ext4 hashes the names — so the listing has to be sorted
+/// before anything picks a file out of it.
+fn sorted_log_files(file_names: impl Iterator<Item = String>) -> Vec<(NaiveDateTime, String)> {
+    let mut files = file_names
+        .filter_map(|name| parse_log_file_ts(&name).map(|ts| (ts, name)))
+        .collect::<Vec<_>>();
+    files.sort();
+    files
+}
+
+/// Every log file but the newest one: that one is still being appended to, every
+/// older one is final.
+fn rotated_log_files(file_names: impl Iterator<Item = String>) -> Vec<(NaiveDateTime, String)> {
+    let mut files = sorted_log_files(file_names);
+    files.pop();
+    files
+}
+
+async fn read_log_file_names(hostname: &str) -> Vec<String> {
    let log_dir = format!("{}/{}/", *TMP_WINDMILL_LOGS_SERVICE, hostname);
-    let rd_dir = tokio::fs::read_dir(log_dir).await;
-    if let Ok(mut log_files) = rd_dir {
-        let mut highest_file: Option<String> = None;
-        let mut second_highest_file: Option<String> = None;
-        while let Ok(Some(file)) = log_files.next_entry().await {
-            let file_name = file
-                .file_name()
-                .to_str()
-                .map(|x| x.to_string())
-                .unwrap_or_default();
-            if file_name > highest_file.clone().unwrap_or_default() {
-                second_highest_file = highest_file;
-                highest_file = Some(file_name);
-            }
+    let mut rd_dir = match tokio::fs::read_dir(&log_dir).await {
+        Ok(rd_dir) => rd_dir,
+        Err(e) => {
+            tracing::error!("Error reading log files: {}, {:#?}", log_dir, e);
+            return vec![];
+        }
+    };
+    let mut file_names = vec![];
+    while let Ok(Some(file)) = rd_dir.next_entry().await {
+        if let Some(file_name) = file.file_name().to_str() {
+            file_names.push(file_name.to_string());
        }
-        (highest_file, second_highest_file)
-    } else {
-        tracing::error!(
-            "Error reading log files: {}, {:#?}",
-            *TMP_WINDMILL_LOGS_SERVICE,
-            rd_dir.unwrap_err()
-        );
-        (None, None)
    }
+    file_names
+}
+
+async fn list_log_files(hostname: &str) -> Vec<(NaiveDateTime, String)> {
+    sorted_log_files(read_log_file_names(hostname).await.into_iter())
+}
+
+async fn list_rotated_log_files(hostname: &str) -> Vec<(NaiveDateTime, String)> {
+    rotated_log_files(read_log_file_names(hostname).await.into_iter())
 }

 fn get_worker_group(mode: &Mode) -> Option<String> {
@@ -1279,133 +1319,188 @@ pub fn send_logs_to_object_store(conn: &Connection, hostname: &str, mode: &Mode)
    tokio::spawn(async move {
        let mut interval = tokio::time::interval(Duration::from_secs(10));
        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+        init_last_log_file_sent(&conn, &hostname).await;
        sleep_until_next_minute_start_plus_one_s().await;
        loop {
            interval.tick().await;
-            let (_, snd_highest_file) = find_two_highest_files(&hostname).await;
-            send_log_file_to_object_store(
-                &hostname,
-                &mode,
-                &worker_group,
-                &conn,
-                snd_highest_file,
-                false,
-            )
-            .await;
+            let files = list_rotated_log_files(&hostname).await;
+            send_log_files_to_object_store(&hostname, &mode, &worker_group, &conn, files).await;
        }
    });
 }

-pub async fn send_current_log_file_to_object_store(conn: &Connection, hostname: &str, mode: &Mode) {
-    tracing::info!("Sending current log file to object store");
-    let (highest_file, _) = find_two_highest_files(hostname).await;
+pub async fn flush_pending_log_files_to_object_store(
+    conn: &Connection,
+    hostname: &str,
+    mode: &Mode,
+) {
+    tracing::info!("Sending pending log files to object store");
    let worker_group = get_worker_group(&mode);
-    send_log_file_to_object_store(hostname, mode, &worker_group, conn, highest_file, true).await;
-}
-
-fn get_now_and_str() -> (NaiveDateTime, String) {
-    let ts = Utc::now().naive_utc();
-    (
-        ts,
-        ts.format(windmill_common::tracing_init::LOG_TIMESTAMP_FMT)
-            .to_string(),
-    )
+    // Nothing rotates after this, so the file still being appended to is registered
+    // here, along with any rotated one the loop had not reached yet. Bounded like the
+    // pool close that follows: a backlog against a slow object store would otherwise
+    // hold the process past its termination grace period. Whatever is left over is
+    // registered by the next run's catch-up.
+    let flush = async {
+        let files = list_log_files(hostname).await;
+        send_log_files_to_object_store(hostname, mode, &worker_group, conn, files).await;
+    };
+    if timeout(Duration::from_secs(15), flush).await.is_err() {
+        tracing::warn!("Could not send all pending log files in time (15s). Exiting anyway.");
+    }
 }

 lazy_static::lazy_static! {
    static ref LAST_LOG_FILE_SENT: Arc<Mutex<Option<NaiveDateTime>>> = Arc::new(Mutex::new(None));
+    /// Serializes the periodic uploader against the shutdown flush. The uploader is a
+    /// detached task that keeps ticking while the flush runs and both walk the same
+    /// files, so without this both can clear the watermark for one file and count its
+    /// lines twice through the additive upsert.
+    static ref SENDING_LOG_FILES: tokio::sync::Mutex<()> = tokio::sync::Mutex::new(());
 }

+fn last_log_file_sent() -> Option<NaiveDateTime> {
+    LAST_LOG_FILE_SENT.lock().ok().and_then(|ts| *ts)
+}
+
+/// Resume from what this host already registered, so a previous run's leftovers reach
+/// the object store rather than being dropped. Their line counts come out zero, this
+/// run having counted none of them, which only flattens their bars in the UI.
+///
+/// The newest registered minute is left out on purpose: the shutdown flush registers
+/// 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 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;
+    };
+    match sqlx::query_scalar!(
+        "SELECT max(log_ts) FROM log_file
+         WHERE hostname = $1 AND log_ts < (SELECT max(log_ts) FROM log_file WHERE hostname = $1)",
+        hostname
+    )
+    .fetch_one(db)
+    .await
+    {
+        Ok(Some(ts)) => {
+            if let Err(e) = LAST_LOG_FILE_SENT.lock().map(|mut last_log_file_sent| {
+                last_log_file_sent.replace(ts);
+            }) {
+                tracing::error!("Error initializing last log file sent: {:?}", e);
+            }
+        }
+        Ok(None) => {}
+        Err(e) => tracing::error!("Error loading last log file sent: {:?}", e),
+    }
+}
+
+async fn send_log_files_to_object_store(
+    hostname: &str,
+    mode: &Mode,
+    worker_group: &Option<String>,
+    conn: &Connection,
+    files: Vec<(NaiveDateTime, String)>,
+) {
+    let _guard = SENDING_LOG_FILES.lock().await;
+    let retention_cutoff = Utc::now().naive_utc()
+        - chrono::Duration::seconds(windmill_common::service_log_retention_secs());
+    for (ts, file_name) in files {
+        if last_log_file_sent().is_some_and(|last| last >= ts) {
+            continue;
+        }
+        // A run coming back from a long outage still finds its predecessor's files on
+        // disk. Registering one past the retention cutoff inserts a row
+        // `delete_expired_items` drops on its next pass, once the indexers have already
+        // paid to parse it.
+        if ts < retention_cutoff {
+            continue;
+        }
+        // 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;
+        }
+    }
+}
+
+/// Returns whether the file ended up registered in `log_file`.
 async fn send_log_file_to_object_store(
    hostname: &str,
    mode: &Mode,
    worker_group: &Option<String>,
    conn: &Connection,
-    snd_highest_file: Option<String>,
-    use_now: bool,
-) {
-    if let Some(highest_file) = snd_highest_file {
-        //parse datetime frome file xxxx.yyyy-MM-dd-HH-mm
-        let (ts, ts_str) = if use_now {
-            get_now_and_str()
-        } else {
-            highest_file
-                .split(".")
-                .last()
-                .and_then(|x| {
-                    NaiveDateTime::parse_from_str(
-                        x,
-                        windmill_common::tracing_init::LOG_TIMESTAMP_FMT,
-                    )
-                    .ok()
-                    .map(|y| (y, x.to_string()))
-                })
-                .unwrap_or_else(get_now_and_str)
-        };
+    file_name: &str,
+    ts: NaiveDateTime,
+) -> bool {
+    #[cfg(feature = "parquet")]
+    if let Some(s3_client) = windmill_object_store::get_object_store().await {
+        let path = std::path::Path::new(&*TMP_WINDMILL_LOGS_SERVICE)
+            .join(hostname)
+            .join(file_name);

-        let exists = LAST_LOG_FILE_SENT.lock().map(|last_log_file_sent| {
-            last_log_file_sent
-                .map(|last_log_file_sent| last_log_file_sent >= ts)
-                .unwrap_or(false)
-        });
-
-        if exists.unwrap_or(false) {
-            return;
-        }
-
-        #[cfg(feature = "parquet")]
-        let s3_client = windmill_object_store::get_object_store().await;
-        #[cfg(feature = "parquet")]
-        if let Some(s3_client) = s3_client {
-            let path = std::path::Path::new(&*TMP_WINDMILL_LOGS_SERVICE)
-                .join(hostname)
-                .join(&highest_file);
-
-            //read file as byte stream
-            let bytes = tokio::fs::read(&path).await;
-            if let Err(e) = bytes {
+        //read file as byte stream
+        let bytes = match tokio::fs::read(&path).await {
+            Ok(bytes) => bytes,
+            Err(e) => {
                tracing::error!("Error reading log file: {:?}", e);
-                return;
+                return false;
            }
-            let path = windmill_object_store::object_store_reexports::Path::from_url_path(format!(
-                "{}{hostname}/{highest_file}",
-                windmill_common::tracing_init::LOGS_SERVICE
-            ));
-            if let Err(e) = path {
+        };
+        let path = windmill_object_store::object_store_reexports::Path::from_url_path(format!(
+            "{}{hostname}/{file_name}",
+            windmill_common::tracing_init::LOGS_SERVICE
+        ));
+        let path = match path {
+            Ok(path) => path,
+            Err(e) => {
                tracing::error!("Error creating log file path: {:?}", e);
-                return;
-            }
-            if let Err(e) = s3_client.put(&path.unwrap(), bytes.unwrap().into()).await {
-                tracing::error!("Error sending logs to object store: {:?}", e);
+                return false;
            }
+        };
+        if let Err(e) = s3_client.put(&path, bytes.into()).await {
+            tracing::error!("Error sending logs to object store: {:?}", e);
+            return false;
        }
+    }

-        let (ok_lines, err_lines) = read_log_counters(ts_str);
+    let ts_str = ts
+        .format(windmill_common::tracing_init::LOG_TIMESTAMP_FMT)
+        .to_string();
+    let (ok_lines, err_lines) = read_log_counters(ts_str);

-        if let Some(db) = conn.as_sql() {
-            match timeout(Duration::from_secs(10), sqlx::query!("INSERT INTO log_file (hostname, mode, worker_group, log_ts, file_path, ok_lines, err_lines, json_fmt)
-             VALUES ($1, $2::text::LOG_MODE, $3, $4, $5, $6, $7, $8)
-             ON CONFLICT (hostname, log_ts) DO UPDATE SET ok_lines = log_file.ok_lines + $6, err_lines = log_file.err_lines + $7",
-                hostname, mode.to_string(), worker_group.clone(), ts, highest_file, ok_lines as i64, err_lines as i64, *JSON_FMT)
-                .execute(db)).await {
-                Ok(Ok(_)) => {
-                    if let Err(e) = LAST_LOG_FILE_SENT.lock().map(|mut last_log_file_sent| {
-                        last_log_file_sent.replace(ts);
-                    }) {
-                        tracing::error!("Error updating last log file sent: {:?}", e);
-                    }
-                    tracing::info!("Log file sent: {}", highest_file);
-                }
-                Ok(Err(e)) => {
-                    tracing::error!("Error inserting log file: {:?}", e);
-                }
-                Err(e) => {
-                    tracing::error!("Error inserting log file, timeout elapsed: {:?}", e);
-                }
+    let Some(db) = conn.as_sql() else {
+        // not sending log file to object store in agent mode
+        return false;
+    };
+
+    match timeout(Duration::from_secs(10), sqlx::query!("INSERT INTO log_file (hostname, mode, worker_group, log_ts, file_path, ok_lines, err_lines, json_fmt)
+     VALUES ($1, $2::text::LOG_MODE, $3, $4, $5, $6, $7, $8)
+     ON CONFLICT (hostname, log_ts) DO UPDATE SET ok_lines = log_file.ok_lines + $6, err_lines = log_file.err_lines + $7",
+        hostname, mode.to_string(), worker_group.clone(), ts, file_name, ok_lines as i64, err_lines as i64, true)
+        .execute(db)).await {
+        Ok(Ok(_)) => {
+            if let Err(e) = LAST_LOG_FILE_SENT.lock().map(|mut last_log_file_sent| {
+                last_log_file_sent.replace(ts);
+            }) {
+                tracing::error!("Error updating last log file sent: {:?}", e);
            }
-        } else {
-            // tracing::warn!("Not sending log file to object store in agent mode");
-            ()
+            tracing::info!("Log file sent: {}", file_name);
+            true
+        }
+        Ok(Err(e)) => {
+            tracing::error!("Error inserting log file: {:?}", e);
+            false
+        }
+        Err(e) => {
+            tracing::error!("Error inserting log file, timeout elapsed: {:?}", e);
+            false
        }
    }
 }
@@ -1593,6 +1688,13 @@ pub async fn trim_resource_versions(db: &DB) -> () {
    }
 }

+/// Matches the batch the settings-page cleanup uses for the same table.
+const SERVICE_LOG_DELETE_BATCH: i64 = 2_000;
+/// Batches per pass. `monitor_db` runs under a 600s timeout that cancels every maintenance
+/// future in the same `join!` and reports a critical error, so a large backlog has to drain
+/// across ticks rather than inside one, the way the neighbouring sweeps already do.
+const SERVICE_LOG_DELETE_MAX_BATCHES: usize = 10;
+
 pub async fn delete_expired_items(db: &DB) -> () {
    let expired_tokens_r = sqlx::query_as!(
        TokenRow,
@@ -1675,23 +1777,48 @@ pub async fn delete_expired_items(db: &DB) -> () {
        Err(e) => tracing::error!("Error deleting cache resource {}", e.to_string()),
    }

-    match sqlx::query_as!(
-        LogFile,
-        "DELETE FROM log_file WHERE log_ts <= now() - ($1::bigint::text || ' s')::interval RETURNING file_path, hostname",
-        SERVICE_LOG_RETENTION_SECS,
-    )
-    .fetch_all(db)
-    .await
-    {
-        Ok(log_files_to_delete) => {
+    // Batched: every process rotates a log file a minute, so lowering the retention makes one
+    // ordinary setting change expire millions of rows at once. An unbounded `DELETE ...
+    // RETURNING` would materialize all of them, and their deletion futures, in this one tick.
+    for _ in 0..SERVICE_LOG_DELETE_MAX_BATCHES {
+        let batch = sqlx::query_as!(
+            LogFile,
+            "DELETE FROM log_file WHERE (hostname, log_ts) IN (
+                 SELECT hostname, log_ts FROM log_file
+                 WHERE log_ts <= now() - ($1::bigint::text || ' s')::interval
+                 LIMIT $2
+             ) RETURNING file_path, hostname",
+            windmill_common::service_log_retention_secs(),
+            SERVICE_LOG_DELETE_BATCH,
+        )
+        .fetch_all(db)
+        .await;
+
+        match batch {
+            Ok(log_files_to_delete) => {
+                if log_files_to_delete.is_empty() {
+                    break;
+                }
+                let n = log_files_to_delete.len();
                let paths = log_files_to_delete
                    .iter()
                    .map(|f| format!("{}/{}", f.hostname, f.file_path))
                    .collect();
-                delete_log_files_from_disk_and_store(paths, &*TMP_WINDMILL_LOGS_SERVICE, windmill_common::tracing_init::LOGS_SERVICE).await;
-
+                delete_log_files_from_disk_and_store(
+                    paths,
+                    &*TMP_WINDMILL_LOGS_SERVICE,
+                    windmill_common::tracing_init::LOGS_SERVICE,
+                )
+                .await;
+                if (n as i64) < SERVICE_LOG_DELETE_BATCH {
+                    break;
+                }
+            }
+            Err(e) => {
+                tracing::error!("Error deleting log file: {:?}", e);
+                break;
+            }
        }
-        Err(e) => tracing::error!("Error deleting log file: {:?}", e),
    }

    let audit_retention_days = audit_log_retention_days().await;
@@ -2798,6 +2925,21 @@ pub async fn reload_retention_period_setting(conn: &Connection) {
    }
 }

+pub async fn reload_service_log_retention_secs_setting(conn: &Connection) {
+    match load_setting_value::<i64>(
+        conn,
+        SERVICE_LOG_RETENTION_SECS_SETTING,
+        "SERVICE_LOG_RETENTION_SECS",
+        DEFAULT_SERVICE_LOG_RETENTION_SECS,
+        |x| x,
+    )
+    .await
+    {
+        Ok(v) => windmill_common::set_service_log_retention_secs(v),
+        Err(e) => tracing::error!("Error reloading service log retention period: {:?}", e),
+    }
+}
+
 pub async fn reload_audit_log_retention_days_setting(conn: &Connection) {
    match load_setting_value::<i64>(
        conn,
@@ -6873,3 +7015,54 @@ mod zombie_worker_memory_pct_tests {
        );
    }
 }
+
+#[cfg(test)]
+mod log_file_listing_tests {
+    use super::{rotated_log_files, sorted_log_files};
+
+    fn names(files: Vec<(chrono::NaiveDateTime, String)>) -> Vec<String> {
+        files.into_iter().map(|(_, n)| n).collect()
+    }
+
+    /// A directory read newest-entry-first is what tmpfs actually hands back.
+    #[test]
+    fn orders_by_minute_whatever_order_readdir_used() {
+        let newest_first = [
+            "h.log.2026-08-29-06-49",
+            "h.log.2026-08-29-06-46",
+            "h.log.2026-08-29-06-48",
+            "h.log.2026-08-29-06-47",
+        ];
+        assert_eq!(
+            names(sorted_log_files(newest_first.iter().map(|x| x.to_string()))),
+            vec![
+                "h.log.2026-08-29-06-46",
+                "h.log.2026-08-29-06-47",
+                "h.log.2026-08-29-06-48",
+                "h.log.2026-08-29-06-49",
+            ]
+        );
+        assert_eq!(
+            names(rotated_log_files(
+                newest_first.iter().map(|x| x.to_string())
+            )),
+            vec![
+                "h.log.2026-08-29-06-46",
+                "h.log.2026-08-29-06-47",
+                "h.log.2026-08-29-06-48",
+            ]
+        );
+    }
+
+    #[test]
+    fn drops_names_that_are_not_rotated_log_files() {
+        let files = sorted_log_files(
+            ["h.log", "not-a-log-file", "h.log.2026-08-29-06-46"]
+                .iter()
+                .map(|x| x.to_string()),
+        );
+        assert_eq!(files.len(), 1);
+        assert_eq!(files[0].1, "h.log.2026-08-29-06-46");
+        assert_eq!(files[0].0.to_string(), "2026-08-29 06:46:00");
+    }
+}
@@ -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)
@@ -308,14 +308,17 @@ async fn test_user_endpoints(db: Pool<Postgres>) -> anyhow::Result<()> {
    let auth_base = format!("http://localhost:{port}/api/auth");

    // --- login (will fail: password hash in fixture is fake) ---
+    // An unparseable stored hash must read as a failed login, not as a server error
+    // relaying the hash parser's message to an unauthenticated caller.
    let resp = client()
        .post(format!("{auth_base}/login"))
        .json(&json!({"email": "test@windmill.dev", "password": "wrong-password"}))
        .send()
        .await
        .unwrap();
-    assert!(
-        resp.status() == 400 || resp.status() == 401 || resp.status() == 500,
+    assert_eq!(
+        resp.status(),
+        400,
        "login: unexpected status {}",
        resp.status()
    );
@@ -804,12 +807,16 @@ async fn test_change_user_email_leaves_group_identities(db: Pool<Postgres>) -> a
    let server = ApiServer::start(db.clone()).await?;
    let global_base = format!("http://localhost:{}/api/users", server.addr.port());

-    sqlx::query!("UPDATE password SET email = 'group-ops@windmill.dev' WHERE email = 'test2@windmill.dev'")
-        .execute(&db)
-        .await?;
-    sqlx::query!("UPDATE usr SET email = 'group-ops@windmill.dev' WHERE email = 'test2@windmill.dev'")
-        .execute(&db)
-        .await?;
+    sqlx::query!(
+        "UPDATE password SET email = 'group-ops@windmill.dev' WHERE email = 'test2@windmill.dev'"
+    )
+    .execute(&db)
+    .await?;
+    sqlx::query!(
+        "UPDATE usr SET email = 'group-ops@windmill.dev' WHERE email = 'test2@windmill.dev'"
+    )
+    .execute(&db)
+    .await?;
    sqlx::query!(
        "INSERT INTO group_(workspace_id, name, summary, extra_perms) VALUES ('test-workspace', 'ops', '', '{}')"
    )
@@ -32,7 +32,7 @@ use windmill_common::tracing_init::{LOGS_SERVICE, TMP_WINDMILL_LOGS_SERVICE};
 use windmill_common::worker::WINDMILL_DIR;
 use windmill_common::{
    DB, INSTANCE_NAME, JOB_RETENTION_SECS, JOB_RETENTION_SECS_OVERRIDES,
-    JOB_RETENTION_SECS_OVERRIDES_LOADED, SERVICE_LOG_RETENTION_SECS,
+    JOB_RETENTION_SECS_OVERRIDES_LOADED,
 };

 use windmill_object_store::object_store_reexports::{
@@ -249,7 +249,7 @@ async fn cleanup_service_logs(
    // Count candidates upfront for progress reporting.
    let total: i64 = sqlx::query_scalar!(
        "SELECT COUNT(*) FROM log_file WHERE log_ts <= now() - ($1::bigint::text || ' s')::interval",
-        SERVICE_LOG_RETENTION_SECS,
+        windmill_common::service_log_retention_secs(),
    )
    .fetch_one(db)
    .await?
@@ -274,7 +274,7 @@ async fn cleanup_service_logs(
                 WHERE log_ts <= now() - ($1::bigint::text || ' s')::interval
                 LIMIT $2
             ) RETURNING file_path, hostname",
-            SERVICE_LOG_RETENTION_SECS,
+            windmill_common::service_log_retention_secs(),
            SERVICE_LOG_BATCH,
        )
        .fetch_all(db)
@@ -680,9 +680,10 @@ async fn cleanup_s3_orphans(
 ) -> error::Result<()> {
    let job_retention_secs = JOB_RETENTION_SECS.load(std::sync::atomic::Ordering::Relaxed);
    let now = Utc::now();
-    // Service logs always have a retention (hardcoded SERVICE_LOG_RETENTION_SECS),
-    // so we scan for service-log orphans regardless of JOB_RETENTION_SECS.
-    let service_cutoff = now - chrono::Duration::seconds(SERVICE_LOG_RETENTION_SECS);
+    // Service logs always have a retention, so we scan for service-log orphans regardless of
+    // JOB_RETENTION_SECS.
+    let service_cutoff =
+        now - chrono::Duration::seconds(windmill_common::service_log_retention_secs());

    // Job-log orphans are only considered once past a job's effective retention window. That window
    // is the instance one OR, for an override workspace (EE), its own — and jobs orphan their logs as
@@ -19,7 +19,7 @@ use windmill_api_auth::ApiAuthed;

 pub use windmill_api_auth::Tokened;

-use argon2::{Argon2, PasswordHash, PasswordVerifier};
+use argon2::{Argon2, PasswordVerifier};
 use axum::{
    extract::{Extension, Path, Query},
    response::{IntoResponse, Response},
@@ -2680,10 +2680,8 @@ async fn login(
    .await?;

    if let Some((email, hash, super_admin)) = email_w_h {
-        let parsed_hash =
-            PasswordHash::new(&hash).map_err(|e| Error::internal_err(e.to_string()))?;
        if argon2
-            .verify_password(password.as_bytes(), &parsed_hash)
+            .verify_password(password.as_bytes(), hash.as_str())
            .is_err()
        {
            audit_log(
@@ -3710,3 +3708,23 @@ async fn request_password_reset(
 }

 // NOTE: reset_password is in windmill-api (depends on users_oss::hash_password EE dispatch)
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Stored hashes outlive the hashing crate: every instance still holds hashes minted by
+    /// older argon2 releases, and an upgrade that stopped reading them locks their users out.
+    #[test]
+    fn verifies_a_hash_minted_by_an_older_argon2() {
+        // The seeded admin hash from migration 20220508150023, m=4096,t=3,p=1.
+        let seeded = "$argon2id$v=19$m=4096,t=3,p=1$oLJo/lPn/gezXCuFOEyaNw$i0T2tCkw3xUFsrBIKZwr8jVNHlIfoxQe+HfDnLtd12I";
+
+        assert!(Argon2::default()
+            .verify_password(b"changeme", seeded)
+            .is_ok());
+        assert!(Argon2::default()
+            .verify_password(b"not-the-password", seeded)
+            .is_err());
+    }
+}
@@ -10,7 +10,7 @@ path = "src/lib.rs"

 [features]
 default = []
-private = ["windmill-audit/private", "windmill-common/private", "windmill-api-auth/private", "windmill-store/private", "windmill-api-users/private", "windmill-api-workspaces/private", "windmill-api-groups/private", "windmill-api-configs/private", "windmill-api-settings/private", "windmill-api-assets/private", "windmill-api-agent-workers?/private", "windmill-trigger-kafka?/private", "windmill-trigger-postgres?/private", "windmill-trigger-mqtt?/private", "windmill-trigger-amqp?/private", "windmill-trigger-websocket?/private", "windmill-trigger-nats?/private", "windmill-trigger-sqs?/private", "windmill-trigger-gcp?/private", "windmill-trigger-azure?/private", "windmill-trigger-email?/private", "windmill-git-sync/private", "windmill-autoscaling?/private", "windmill-object-store/private", "windmill-api-npm-proxy/private"]
+private = ["windmill-audit/private", "windmill-common/private", "windmill-api-auth/private", "windmill-store/private", "windmill-api-users/private", "windmill-api-workspaces/private", "windmill-api-groups/private", "windmill-api-configs/private", "windmill-api-settings/private", "windmill-api-assets/private", "windmill-api-agent-workers?/private", "windmill-trigger-kafka?/private", "windmill-trigger-postgres?/private", "windmill-trigger-mqtt?/private", "windmill-trigger-amqp?/private", "windmill-trigger-websocket?/private", "windmill-trigger-nats?/private", "windmill-trigger-sqs?/private", "windmill-trigger-gcp?/private", "windmill-trigger-azure?/private", "windmill-trigger-email?/private", "windmill-git-sync/private", "windmill-autoscaling?/private", "windmill-object-store/private", "windmill-api-npm-proxy/private", "windmill-indexer?/private"]
 enterprise = ["windmill-queue/enterprise", "windmill-audit/enterprise", "windmill-git-sync/enterprise", "windmill-common/enterprise", "windmill-worker?/enterprise", "windmill-api-auth/enterprise", "windmill-store/enterprise", "windmill-api-jobs/enterprise", "windmill-api-scripts/enterprise", "windmill-api-flows/enterprise", "windmill-api-users/enterprise", "windmill-api-workspaces/enterprise", "windmill-api-groups/enterprise", "windmill-api-configs/enterprise", "windmill-api-settings/enterprise", "windmill-api-schedule/enterprise", "windmill-api-debug/enterprise", "windmill-api-agent-workers?/enterprise", "windmill-trigger/enterprise", "windmill-trigger-kafka?/enterprise", "windmill-trigger-postgres?/enterprise", "windmill-trigger-mqtt?/enterprise", "windmill-trigger-amqp?/enterprise", "windmill-trigger-websocket?/enterprise", "windmill-trigger-email?/enterprise", "windmill-trigger-nats?/enterprise", "windmill-trigger-sqs?/enterprise", "windmill-trigger-gcp?/enterprise", "windmill-trigger-azure?/enterprise", "windmill-trigger-http?/enterprise", "windmill-native-triggers?/enterprise", "dep:windmill-autoscaling", "windmill-autoscaling/enterprise", "windmill-api-npm-proxy/enterprise", "license"]
 stripe = []
 run_inline = ["dep:windmill-worker", "windmill-api-configs/run_inline"]
@@ -18,10 +18,12 @@ agent_worker_server = ["dep:windmill-worker", "dep:windmill-api-agent-workers"]
 enterprise_saml = ["dep:samael", "dep:libxml"]
 benchmark = []
 embedding = ["windmill-api-embeddings/embedding"]
-parquet = ["dep:datafusion", "windmill-common/parquet", "windmill-object-store/parquet", "windmill-worker?/parquet", "windmill-api-users/parquet", "windmill-api-settings/parquet", "windmill-api-workspaces/parquet", "windmill-api-npm-proxy/parquet", "dep:aws-sigv4", "dep:aws-sdk-config", "dep:quick-xml"]
+parquet = ["dep:datafusion", "windmill-common/parquet", "windmill-object-store/parquet", "windmill-worker?/parquet", "windmill-indexer?/parquet", "windmill-api-users/parquet", "windmill-api-settings/parquet", "windmill-api-workspaces/parquet", "windmill-api-npm-proxy/parquet", "dep:aws-sigv4", "dep:aws-sdk-config", "dep:quick-xml"]
 prometheus = ["windmill-common/prometheus", "windmill-queue/prometheus", "dep:prometheus", "windmill-worker?/prometheus", "windmill-api-scripts/prometheus"]
 openidconnect = ["dep:openidconnect", "windmill-common/openidconnect", "windmill-store/openidconnect"]
-tantivy = ["dep:windmill-indexer"]
+# The service log search handler reads the columnar store, so tantivy alone is
+# not enough for this crate to build on its own.
+tantivy = ["dep:windmill-indexer", "parquet"]
 kafka = ["dep:windmill-trigger-kafka", "windmill-store/kafka"]
 kafka-gssapi = ["kafka", "windmill-trigger-kafka/kafka-gssapi"]
 nats = ["dep:windmill-trigger-nats", "windmill-store/nats"]
@@ -23890,7 +23890,7 @@ paths:
                    items:
                      type: string
                  hits:
-                    description: log files that matched the query
+                    description: the log lines that matched the query, newest first
                    type: array
                    items:
                      $ref: "#/components/schemas/LogSearchHit"
@@ -34355,8 +34355,34 @@ components:
    LogSearchHit:
      type: object
      properties:
-        dancer:
+        ts:
+          description: timestamp of the log line itself, not of the file containing it
          type: string
+          format: date-time
+        host:
+          type: string
+        level:
+          type: string
+          enum: [TRACE, DEBUG, INFO, WARN, ERROR]
+        target:
+          description: the tracing target that emitted the line
+          type: string
+          nullable: true
+        message:
+          type: string
+        file_path:
+          description: the log file the line came from
+          type: string
+        line_no:
+          description: offset of the line within its file
+          type: integer
+      required:
+        - ts
+        - host
+        - level
+        - message
+        - file_path
+        - line_no

    AutoscalingEvent:
      type: object
@@ -88,6 +88,66 @@ async fn list_files(
    Ok(Json(rows))
 }

+/// Rebuild one source log file from the columnar store.
+///
+/// Not the original bytes: the store holds a line's fields rather than its text,
+/// so the JSON is re-serialized here and key order and whitespace are this
+/// writer's. Everything a reader can see survives — the drawer this feeds
+/// renders a prettified view of each line either way, and a line that was never
+/// JSON comes back exactly as it was written.
+#[cfg(all(feature = "tantivy", feature = "private"))]
+async fn get_log_file_from_store(
+    db: &DB,
+    store: &windmill_indexer::service_logs_store_ee::Store,
+    path: &str,
+) -> windmill_common::error::Result<Response> {
+    let (hostname, file_name) = path
+        .split_once('/')
+        .ok_or_else(|| Error::BadRequest("Invalid path".to_string()))?;
+
+    // The store is partitioned by day and mode, neither of which the path
+    // carries. `log_file` names both, and its primary key starts with hostname.
+    let file = sqlx::query!(
+        // `mode!` because the column is NOT NULL and only the cast makes sqlx
+        // think otherwise; a silent default would look up a `mode=` partition
+        // that matches nothing and read as a missing file.
+        "SELECT mode::text AS \"mode!\", log_ts FROM log_file WHERE hostname = $1 AND file_path = $2 ORDER BY log_ts DESC LIMIT 1",
+        hostname,
+        file_name
+    )
+    .fetch_optional(db)
+    .await?
+    .ok_or_else(|| Error::NotFound(format!("File {path} not found")))?;
+
+    // A row registered by this version carries the minute in the file's own name,
+    // so the two agree and the second is redundant. One written before the
+    // uploader derived `log_ts` from the name carries a wall clock instead, and
+    // those outlive an upgrade by the retention period — which is also what makes
+    // the `ORDER BY` above worth having. The name is authoritative, so both go.
+    let mut known_ts = vec![chrono::DateTime::from_naive_utc_and_offset(
+        file.log_ts,
+        chrono::Utc,
+    )];
+    if let Some(named) = file_name.rsplit('.').next().and_then(|s| {
+        chrono::NaiveDateTime::parse_from_str(s, windmill_common::tracing_init::LOG_TIMESTAMP_FMT)
+            .ok()
+    }) {
+        known_ts.push(chrono::DateTime::from_naive_utc_and_offset(
+            named,
+            chrono::Utc,
+        ));
+    }
+
+    let text = windmill_indexer::service_logs_store_ee::read_log_file(
+        store, &file.mode, hostname, file_name, &known_ts,
+    )
+    .await
+    .map_err(|e| Error::internal_err(format!("Error reading the service log store: {e}")))?
+    .ok_or_else(|| Error::NotFound(format!("File {path} not found")))?;
+
+    Ok(content_plain(Body::from(text)))
+}
+
 async fn get_log_file(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
@@ -104,27 +164,30 @@ async fn get_log_file(
    let s3_client = windmill_object_store::get_object_store().await;
    #[cfg(feature = "parquet")]
    if let Some(s3_client) = s3_client {
-        let path = format!("{}{}", windmill_common::tracing_init::LOGS_SERVICE, path);
-        let file = s3_client
+        use windmill_object_store::object_store_reexports::ObjectStoreError;
+
+        // The raw file, for as long as it is there. It outlives its ingestion by
+        // one indexer pass at most, so this covers the most recent minutes of a
+        // host's logs byte for byte; everything older is rebuilt from the store.
+        let object_path = format!("{}{}", windmill_common::tracing_init::LOGS_SERVICE, path);
+        match s3_client
            .get(&windmill_object_store::object_store_reexports::Path::from(
-                path,
+                object_path,
            ))
-            .await;
-        match file {
-            Ok(file) => {
-                let bytes = file.bytes().await;
-                match bytes {
-                    Ok(bytes) => {
-                        return Ok(content_plain(Body::from(bytes::Bytes::from(bytes))));
-                    }
-                    Err(e) => {
-                        return Err(Error::internal_err(format!(
-                            "Error pulling the bytes: {}",
-                            e
-                        )));
-                    }
+            .await
+        {
+            Ok(file) => match file.bytes().await {
+                Ok(bytes) => {
+                    return Ok(content_plain(Body::from(bytes::Bytes::from(bytes))));
                }
-            }
+                Err(e) => {
+                    return Err(Error::internal_err(format!(
+                        "Error pulling the bytes: {}",
+                        e
+                    )));
+                }
+            },
+            Err(ObjectStoreError::NotFound { .. }) => {}
            Err(e) => {
                return Err(Error::internal_err(format!(
                    "Error fetching the file: {}",
@@ -132,6 +195,11 @@ async fn get_log_file(
                )));
            }
        }
+
+        #[cfg(all(feature = "tantivy", feature = "private"))]
+        return get_log_file_from_store(&db, &s3_client, &path).await;
+        #[cfg(not(all(feature = "tantivy", feature = "private")))]
+        return Err(Error::NotFound(format!("File {path} not found")));
    }
    let full_path = format!("{}{}", *TMP_WINDMILL_LOGS_SERVICE, path);
    // SECURITY (defense in depth): refuse to read through a symlink so a planted
@@ -15,6 +15,7 @@ pub const OAUTH_SETTING: &str = "oauths";
 pub const AI_CONFIG_SETTING: &str = "ai_config";
 pub const RETENTION_PERIOD_SECS_SETTING: &str = "retention_period_secs";
 pub const RETENTION_PERIOD_SECS_OVERRIDES_SETTING: &str = "retention_period_secs_overrides";
+pub const SERVICE_LOG_RETENTION_SECS_SETTING: &str = "service_log_retention_secs";
 /// Upper bound on how many per-workspace retention overrides may be configured. The periodic monitor
 /// sweeps each override workspace in its own transaction every pass, so this keeps a pass bounded
 /// (and the feature is a targeted escape hatch for a handful of special workspaces, not a bulk knob).
@@ -94,6 +94,21 @@ pub async fn load_indexer_config(db: &DB) -> error::Result<TantivyIndexerSetting
    })
 }

+/// How far back the service log index reaches, in seconds.
+///
+/// [`crate::service_log_retention_secs`] is the ceiling: past it a line's `log_file` row is
+/// deleted and can no longer be indexed. `max_index_time_window_secs` of `0` means "do not
+/// shrink below that ceiling", not "unbounded" — both sites that trim and populate the index
+/// derive the window here so the two cannot disagree about it.
+pub fn service_log_index_window_secs(max_index_time_window_secs: i64) -> i64 {
+    let retention = crate::service_log_retention_secs();
+    if max_index_time_window_secs > 0 {
+        std::cmp::min(max_index_time_window_secs, retention)
+    } else {
+        retention
+    }
+}
+
 pub fn get_env_var(env_var: &str) -> Option<u64> {
    match std::env::var(env_var).map(|x| x.parse()) {
        Ok(Ok(i)) => Some(i),
@@ -136,3 +151,60 @@ pub fn get_indexer_rates_from_env() -> TantivyIndexerSettings {

    settings
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // One test rather than several: both halves share the process-wide retention, and the
+    // setter half writes it, which parallel tests would race.
+    #[test]
+    fn retention_rejects_unusable_values_and_the_index_window_clamps_to_it() {
+        use crate::{
+            service_log_retention_secs, set_service_log_retention_secs,
+            DEFAULT_SERVICE_LOG_RETENTION_SECS,
+        };
+
+        // See `set_service_log_retention_secs` for why the two unusable directions land apart:
+        // too large keeps the intent by capping, non-positive cannot and falls back.
+        let rejected: Vec<i64> = [0, -1, i64::MIN]
+            .iter()
+            .map(|v| {
+                set_service_log_retention_secs(*v);
+                service_log_retention_secs()
+            })
+            .collect();
+        let capped: Vec<i64> = [i64::MAX, 60 * 60 * 24 * 365 * 101]
+            .iter()
+            .map(|v| {
+                set_service_log_retention_secs(*v);
+                service_log_retention_secs()
+            })
+            .collect();
+
+        set_service_log_retention_secs(60 * 60 * 24 * 3);
+        let retention = service_log_retention_secs();
+        let windows = [
+            // `0` disables the extra shrinking rather than lifting the ceiling — the trap that
+            // makes an unset setting look unbounded.
+            service_log_index_window_secs(0),
+            // Retention is the ceiling: the index cannot reach lines whose `log_file` row is gone.
+            service_log_index_window_secs(retention * 2),
+            service_log_index_window_secs(60),
+        ];
+        set_service_log_retention_secs(DEFAULT_SERVICE_LOG_RETENTION_SECS);
+
+        assert_eq!(
+            rejected,
+            vec![DEFAULT_SERVICE_LOG_RETENTION_SECS; 3],
+            "a value that would expire everything must fall back to the default"
+        );
+        assert_eq!(
+            capped,
+            vec![60 * 60 * 24 * 365 * 100; 2],
+            "an oversized value must cap, not shorten retention to the default"
+        );
+        assert_eq!(retention, 60 * 60 * 24 * 3);
+        assert_eq!(windows, [retention, retention, 60]);
+    }
+}
@@ -147,9 +147,53 @@ pub const DEFAULT_MAX_CONNECTIONS_INDEXER: u32 = 5;

 pub const DEFAULT_HUB_BASE_URL: &str = "https://hub.windmill.dev";
 pub const PRIVATE_HUB_MIN_VERSION: i32 = 10_000_000;
-pub const SERVICE_LOG_RETENTION_SECS: i64 = 60 * 60 * 24 * 14; // 2 weeks retention period for logs
+pub const DEFAULT_SERVICE_LOG_RETENTION_SECS: i64 = 60 * 60 * 24 * 14; // 2 weeks retention period for logs
 pub const WM_DEPLOYERS_GROUP: &str = "wm_deployers";

+/// A century. Every consumer has to survive `now - retention`, and the ceilings are much lower
+/// than an `i64`: `DateTime` subtraction panics past year 262143, and the `(<n> s)::interval`
+/// the cleanup queries build overflows Postgres' microsecond field.
+const MAX_SERVICE_LOG_RETENTION_SECS: i64 = 60 * 60 * 24 * 365 * 100;
+
+/// Apply a configured service log retention, in seconds.
+///
+/// The only way into [`SERVICE_LOG_RETENTION_SECS`], so an unusable value can never reach a
+/// cutoff. The two unusable directions are not the same mistake and must not share a landing
+/// point: too large still says "keep these for a very long time", so it is capped and the
+/// intent survives, whereas falling back would delete logs the operator meant to keep. A
+/// non-positive value has no such reading — every cutoff is `now - retention`, so it lands at
+/// or after `now` and the next sweep expires the entire history, rows and object-storage files
+/// alike. Unlike job retention there is no "keep forever" spelling here, so `0` — what an
+/// operator types by analogy with it, and what the settings UI writes into a field that was
+/// merely focused — falls back to the default.
+pub fn set_service_log_retention_secs(configured: i64) {
+    let effective = if configured > MAX_SERVICE_LOG_RETENTION_SECS {
+        tracing::warn!(
+            "service log retention of {configured}s exceeds the maximum of \
+             {MAX_SERVICE_LOG_RETENTION_SECS}s, capping it there"
+        );
+        MAX_SERVICE_LOG_RETENTION_SECS
+    } else if configured >= 1 {
+        configured
+    } else {
+        tracing::warn!(
+            "service log retention of {configured}s would expire every service log, \
+             falling back to the default of {DEFAULT_SERVICE_LOG_RETENTION_SECS}s"
+        );
+        DEFAULT_SERVICE_LOG_RETENTION_SECS
+    };
+    SERVICE_LOG_RETENTION_SECS.store(effective, std::sync::atomic::Ordering::Relaxed);
+}
+
+/// How long a service log line stays retrievable, in seconds.
+///
+/// The outer bound on everything service-log: the `log_file` rows, the raw files in object
+/// storage, the columnar store queried by retrieval, and — through
+/// [`indexer::service_log_index_window_secs`] — the search index.
+pub fn service_log_retention_secs() -> i64 {
+    SERVICE_LOG_RETENTION_SECS.load(std::sync::atomic::Ordering::Relaxed)
+}
+
 /// Canonical form of a base URL, used as one of the inputs to the offline-license
 /// instance hash (`compute_instance_hash`).
 ///
@@ -375,6 +419,10 @@ lazy_static::lazy_static! {
    /// workspace configured before its override could be read.
    pub static ref JOB_RETENTION_SECS_OVERRIDES_LOADED: AtomicBool = AtomicBool::new(false);
    pub static ref AUDIT_LOG_RETENTION_DAYS: AtomicI64 = AtomicI64::new(0);
+    /// Private on purpose: [`set_service_log_retention_secs`] is the only writer, so a value that
+    /// would expire every service log cannot reach a cutoff. Read it with
+    /// [`service_log_retention_secs`].
+    static ref SERVICE_LOG_RETENTION_SECS: AtomicI64 = AtomicI64::new(DEFAULT_SERVICE_LOG_RETENTION_SECS);

    pub static ref MONITOR_LOGS_ON_OBJECT_STORE: AtomicBool = AtomicBool::new(false);

@@ -178,56 +178,40 @@ pub fn initialize_tracing(
        .with(logs_bridge.with_filter(otel_logs_filter))
        .with(opentelemetry_filtered);

-    match *JSON_FMT {
-        true => {
-            // Stdout layer with its own filter
-            let stdout_layer = json_layer()
-                .with_writer(std::io::stdout)
-                .flatten_event(true)
-                .with_filter(stdout_env_filter)
-                .with_filter(create_targets_filter(default_env_filter));
+    // The service log files are written to be indexed, not tailed, so they always carry the
+    // JSON format: it is what preserves level, target and the current span as fields rather
+    // than as text the index would have to recover by regex. JSON_FMT governs stdout only.
+    let file_layer = json_layer()
+        .with_writer(log_file_writer)
+        .flatten_event(true)
+        .with_filter(file_env_filter)
+        .with_filter(create_targets_filter(default_env_filter));

-            // File layer with its own filter
-            let file_layer = json_layer()
-                .with_writer(log_file_writer)
-                .flatten_event(true)
-                .with_filter(file_env_filter)
-                .with_filter(create_targets_filter(default_env_filter));
+    // Boxed so both arms have one type: the file layer is a single value and could not
+    // otherwise be typed against two different subscriber stacks.
+    let stdout_layer = match *JSON_FMT {
+        true => json_layer()
+            .with_writer(std::io::stdout)
+            .flatten_event(true)
+            .with_filter(stdout_env_filter)
+            .with_filter(create_targets_filter(default_env_filter))
+            .boxed(),
+        false => compact_layer()
+            .with_writer(std::io::stdout)
+            .with_ansi(style.to_lowercase() != "never")
+            .with_file(true)
+            .with_line_number(true)
+            .with_target(false)
+            .with_filter(stdout_env_filter)
+            .with_filter(create_targets_filter(default_env_filter))
+            .boxed(),
+    };

-            base_layer
-                .with(stdout_layer)
-                .with(file_layer)
-                .with(CountingLayer::new())
-                .init()
-        }
-        false => {
-            // Stdout layer with its own filter
-            let stdout_layer = compact_layer()
-                .with_writer(std::io::stdout)
-                .with_ansi(style.to_lowercase() != "never")
-                .with_file(true)
-                .with_line_number(true)
-                .with_target(false)
-                .with_filter(stdout_env_filter)
-                .with_filter(create_targets_filter(default_env_filter));
-
-            // File layer with its own filter
-            let file_layer = compact_layer()
-                .with_writer(log_file_writer)
-                .with_ansi(false) // No ANSI codes in log files
-                .with_file(true)
-                .with_line_number(true)
-                .with_target(false)
-                .with_filter(file_env_filter)
-                .with_filter(create_targets_filter(default_env_filter));
-
-            base_layer
-                .with(stdout_layer)
-                .with(file_layer)
-                .with(CountingLayer::new())
-                .init()
-        }
-    }
+    base_layer
+        .with(stdout_layer)
+        .with(file_layer)
+        .with(CountingLayer::new())
+        .init();
    (_guard, meter_provider)
 }

@@ -10,7 +10,13 @@ path = "src/lib.rs"

 [features]
 default = []
-parquet = ["windmill-common/parquet", "windmill-object-store/parquet"]
+parquet = [
+    "windmill-common/parquet",
+    "windmill-object-store/parquet",
+    "dep:datafusion",
+    "dep:object_store",
+    "dep:url",
+]
 private = ["windmill-common/private"]
 enterprise = ["windmill-common/enterprise", "windmill-object-store/enterprise"]

@@ -33,3 +39,6 @@ astral-tokio-tar.workspace = true
 lazy_static.workspace = true
 const_format.workspace = true
 flume.workspace = true
+datafusion = { workspace = true, optional = true }
+object_store = { workspace = true, optional = true }
+url = { workspace = true, optional = true }
@@ -7,3 +7,5 @@ pub mod indexer_oss;
 #[cfg(feature = "private")]
 pub mod service_logs_ee;
 pub mod service_logs_oss;
+#[cfg(all(feature = "private", feature = "parquet"))]
+pub mod service_logs_store_ee;
@@ -1143,6 +1143,32 @@
 				description="Configure default timeouts and retention policies for job execution."
 				link="https://www.windmill.dev/docs/advanced/instance_settings#jobs"
 			/>
+		{:else if category == 'Service logs'}
+			<SettingsPageHeader
+				title="Service logs"
+				description="The logs of the Windmill processes themselves — servers, workers and the indexer. Job logs are covered by the job retention period under Jobs."
+			/>
+			{#if !$values['object_store_cache_config']}
+				<div class="pb-4">
+					<Alert type="info" title="Log files stay on local disk" size="xs">
+						Instance object storage is not configured, so every server and worker keeps its log
+						files on its own disk. This page lists what each host wrote, but can only open the
+						files belonging to the replica serving the request — another host's are listed and
+						not readable — and a host's files go with it when it is replaced. Retention below
+						still governs the entries in the database and the files on disk.
+					</Alert>
+				</div>
+			{:else if !$enterpriseLicense}
+				<div class="pb-4">
+					<Alert type="info" title="Raw log files accumulate without the indexer" size="xs">
+						Log files are uploaded to instance object storage, and the indexer that would ingest
+						them into the columnar store and delete each one afterwards is an enterprise
+						feature. Retention below expires the database entries and the local files; the
+						uploaded copies are only removed when <b>Delete logs from s3 periodically</b> is on
+						under Object Storage.
+					</Alert>
+				</div>
+			{/if}
 		{:else if category == 'Object Storage'}
 			<SettingsPageHeader
 				title="Object Storage"
@@ -2,7 +2,7 @@
 	import { createBubbler, preventDefault } from 'svelte/legacy'

 	const bubble = createBubbler()
-	import { IndexSearchService, ServiceLogsService } from '$lib/gen'
+	import { IndexSearchService, ServiceLogsService, type LogSearchHit } from '$lib/gen'

 	import TimeframeSelect, {
 		serviceLogsTimeframes,
@@ -253,39 +253,50 @@
 		try {
 			let res = ''
 			log.split('\n').forEach((line) => {
+				// A file can hold both formats: the ones written before the layer
+				// switched to JSON, and panics or subprocess output that was never
+				// JSON to begin with. Those lines pass through as they are rather
+				// than being dropped, which would render the file blank.
+				let obj: any = undefined
 				if (line.startsWith('{') && line.endsWith('}')) {
-					let obj = JSON.parse(line)
-					if (typeof obj == 'object') {
-						let nl = ''
-						if (obj['timestamp']) {
-							nl += obj['timestamp'] + ' '
-						}
-						if (obj['level']) {
-							let lvl = obj['level']
-							if (lvl == 'ERROR') {
-								nl += '\x1b[31mERROR\x1b[0m '
-							} else if (lvl == 'INFO') {
-								nl += '\x1b[32mINFO\x1b[0m '
-							} else {
-								nl += obj['level'] + ' '
-							}
-						}
-						if (obj['message']) {
-							nl += obj['message'] + ' '
-						}
-						delete obj['timestamp']
-						delete obj['level']
-						delete obj['message']
-						Object.keys(obj).forEach((key) => {
-							nl +=
-								key +
-								'=' +
-								(typeof obj[key] == 'object' ? JSON.stringify(obj[key]) : obj[key]) +
-								' '
-						})
-						res += nl + '\n'
+					try {
+						obj = JSON.parse(line)
+					} catch {
+						obj = undefined
 					}
 				}
+				if (obj === null || typeof obj !== 'object') {
+					res += line + '\n'
+				} else {
+					let nl = ''
+					if (obj['timestamp']) {
+						nl += obj['timestamp'] + ' '
+					}
+					if (obj['level']) {
+						let lvl = obj['level']
+						if (lvl == 'ERROR') {
+							nl += '\x1b[31mERROR\x1b[0m '
+						} else if (lvl == 'INFO') {
+							nl += '\x1b[32mINFO\x1b[0m '
+						} else {
+							nl += obj['level'] + ' '
+						}
+					}
+					if (obj['message']) {
+						nl += obj['message'] + ' '
+					}
+					delete obj['timestamp']
+					delete obj['level']
+					delete obj['message']
+					Object.keys(obj).forEach((key) => {
+						nl +=
+							key +
+							'=' +
+							(typeof obj[key] == 'object' ? JSON.stringify(obj[key]) : obj[key]) +
+							' '
+					})
+					res += nl + '\n'
+				}
 			})

 			return res
@@ -294,6 +305,23 @@
 		}
 	}

+	// A hit is one log line with its fields already separated, so rendering it is
+	// formatting rather than parsing — there is no JSON to prettify and no
+	// snippet to highlight.
+	function renderHit(hit: LogSearchHit): string {
+		const level =
+			hit.level === 'ERROR'
+				? '\x1b[31mERROR\x1b[0m'
+				: hit.level === 'WARN'
+					? '\x1b[33mWARN\x1b[0m'
+					: hit.level === 'INFO'
+						? '\x1b[32mINFO\x1b[0m'
+						: hit.level
+		return [hit.ts, level, hit.message, hit.target ? `target=${hit.target}` : '']
+			.filter(Boolean)
+			.join(' ')
+	}
+
 	let logs: any = $state()

 	let debounceTimeout: number | undefined = undefined
@@ -399,7 +427,9 @@
 	) {
 		const res = await ServiceLogsService.getLogFile({ path: `${hostname}/${path}` })

-		content = processLogWithJsonFmt(ansi_up.ansi_to_html(res), jsonFmt)
+		// Prettify first: it emits its own ANSI for the level, which converting
+		// beforehand would leave in the output as literal escapes.
+		content = ansi_up.ansi_to_html(processLogWithJsonFmt(res, jsonFmt))
 		hitLineNumber = lineNumber
 		logDrawerOpen = true

@@ -687,23 +717,19 @@
 						</div>
 					{:else if logs != undefined}
 						<div class="flex flex-col min-w-full w-fit">
-							{#each logs.hits as { snippet_fragment, snippet_highlighted, document }}
+							<!-- Keyed: LogSnippetViewer renders its html once at creation, so an
+										 index-reused instance would keep the previous search's line. -->
+							{#each logs.hits ?? [] as hit, i (`${i}:${hit.file_path}:${hit.line_no}`)}
 								<LogSnippetViewer
-									content={snippet_fragment || document.logs[0]}
-									highlighted={snippet_highlighted}
-									onClick={() => {
-										let logLineNumber = document.line_number[0]
-										let logFile = document.file_name[0]
-										let host = document.host[0]
-										let jsonFmt = document.json_fmt[0]
-										seeLogContext(logLineNumber, logFile, host, jsonFmt)
-									}}
+									content={renderHit(hit)}
+									highlighted={[]}
+									onClick={() => seeLogContext(hit.line_no, hit.file_path, hit.host, true)}
 								/>
 							{/each}
-							{#if logs.hits.length === 0}
+							{#if (logs.hits ?? []).length === 0}
 								<div class="text-center py-20 text-bold text-xl text-primary"> No logs </div>
 							{/if}
-							{#if logs.hits.length === 1000}
+							{#if (logs.hits ?? []).length === 1000}
 								<div class="pl-6 py-6 text-sm text-secondary">
 									Older matches were truncated from this search, try refining your filters to get
 									more precise results.
@@ -999,6 +999,23 @@ export const settings: Record<string, Setting[]> = {
 			triggersRestart: true
 		}
 	],
+	'Service logs': [
+		{
+			label: 'Retention in secs',
+			key: 'service_log_retention_secs',
+			description:
+				'How long a service log is kept, across every copy of it: the entry in the database, the file on the disk of the process that wrote it, and — once instance object storage is configured and the indexer has ingested it — its line in the columnar store that search and the log viewer read. Search reaches back at most this far, and less when the indexer time window under Indexer is shorter. Defaults to 14 days. There is no keep-forever setting here — leave it empty for the default.',
+			fieldType: 'seconds',
+			storage: 'setting',
+			cloudonly: false,
+			error:
+				'Service log retention must be between 1 second and 100 years — leave it empty for the default',
+			isValid: (value: any) =>
+				value == undefined ||
+				(typeof value === 'number' && value > 0 && value <= 60 * 60 * 24 * 365 * 100)
+		}
+	],
+
 	Indexer: [
 		{
 			label: '',
@@ -1190,6 +1207,12 @@ export const instanceSettingsNavigationGroups = [
 				aiDescription: 'Instance OTEL/Prometheus settings',
 				isEE: true
 			},
+			{
+				id: 'service_logs',
+				label: 'Service logs',
+				aiId: 'instance-settings-service-logs',
+				aiDescription: 'Service log retention settings'
+			},
 			{
 				id: 'indexer',
 				label: 'Indexer',
@@ -1273,6 +1296,7 @@ export const tabToCategoryMap: Record<string, string> = {
 	webhooks: 'Webhooks',
 	otel_prom: 'OTEL/Prom',
 	indexer: 'Indexer',
+	service_logs: 'Service logs',
 	telemetry: 'Telemetry',
 	secret_storage: 'Secret Storage',
 	object_storage: 'Object Storage',
@@ -1308,6 +1332,7 @@ export const categoryToTabMap: Record<string, string> = {
 	Webhooks: 'webhooks',
 	'OTEL/Prom': 'otel_prom',
 	Indexer: 'indexer',
+	'Service logs': 'service_logs',
 	Telemetry: 'telemetry',
 	'Secret Storage': 'secret_storage',
 	'Object Storage': 'object_storage',