From 25c50d2e03b7910eb044caa0e87d29dac42dc5df Mon Sep 17 00:00:00 2001
From: udev <uddhav@openobserve.ai>
Date: Wed, 17 Sep 2025 14:24:14 +0530
Subject: [PATCH] fix: update remote wal config (#8359)

### **PR Type**
Enhancement, Bug fix


___

### **Description**
- Add two new pipeline config options

- Set safe defaults when zero provided

- Improve WAL pipeline timing controls


___

### Diagram Walkthrough


```mermaid
flowchart LR
  CFG["Config struct (Pipeline)"] -- "add fields" --> F1["pipeline_file_push_back_interval (secs)"]
  CFG -- "add fields" --> F2["pipeline_sink_task_spawn_interval_ms (ms)"]
  VALID["check_pipeline_config"] -- "apply defaults if 0" --> F1
  VALID -- "apply defaults if 0" --> F2
```



<details> <summary><h3> File Walkthrough</h3></summary>

<table><thead><tr><th></th><th align="left">Relevant
files</th></tr></thead><tbody><tr><td><strong>Enhancement</strong></td><td><table>
<tr>
  <td>
    <details>
<summary><strong>config.rs</strong><dd><code>New pipeline timing fields
and defaulting logic</code>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; &nbsp; &nbsp; </dd></summary>
<hr>

src/config/src/config.rs

<ul><li>Add <code>pipeline_file_push_back_interval</code> with env
config.<br> <li> Add <code>pipeline_sink_task_spawn_interval_ms</code>
with env config.<br> <li> Enforce non-zero defaults in
<code>check_pipeline_config</code>.<br> <li> Preserve existing WAL size
handling.</ul>


</details>


  </td>
<td><a
href="https://github.com/openobserve/openobserve/pull/8359/files#diff-9aeee4a74c0520daa40f9b81c7016f5b6b3254375b968dd7598e745630980d62">+20/-0</a>&nbsp;
&nbsp; </td>

</tr>
</table></td></tr></tr></tbody></table>

</details>

___
---
 src/config/src/config.rs | 20 ++++++++++++++++++++
 1 file changed, 20 insertions(+)

diff --git a/src/config/src/config.rs b/src/config/src/config.rs
index ace9006b9a..c722b4bf6d 100644
--- a/src/config/src/config.rs
+++ b/src/config/src/config.rs
@@ -1936,6 +1936,18 @@ pub struct Pipeline {
         help = "pipeline max file retention time in seconds"
     )]
     pub pipeline_max_file_retention_time_seconds: u64,
+    #[env_config(
+        name = "ZO_PIPELINE_FILE_PUSH_BACK_INTERVAL",
+        default = 2,
+        help = "duration in seconds to push the file to back to the queue after a read complete"
+    )]
+    pub pipeline_file_push_back_interval: u64,
+    #[env_config(
+        name = "ZO_PIPELINE_SINK_TASK_SPAWN_INTERVAL_MS",
+        default = 100,
+        help = "interval in milliseconds to spawn a new sink task"
+    )]
+    pub pipeline_sink_task_spawn_interval_ms: u64,
 }
 
 #[derive(EnvConfig, Default)]
@@ -2781,6 +2793,14 @@ fn check_pipeline_config(cfg: &mut Config) -> Result<(), anyhow::Error> {
     } else {
         cfg.pipeline.wal_size_limit *= 1024 * 1024;
     }
+
+    if cfg.pipeline.pipeline_file_push_back_interval == 0 {
+        cfg.pipeline.pipeline_file_push_back_interval = 2; // 2 seconds
+    }
+
+    if cfg.pipeline.pipeline_sink_task_spawn_interval_ms == 0 {
+        cfg.pipeline.pipeline_sink_task_spawn_interval_ms = 100; // 100 milliseconds
+    }
     Ok(())
 }