platform_core/util/
elastic_queue.rs

1//
2// Copyright 2018-2026 Accenture Technology
3//
4// Licensed under the Apache License, Version 2.0 (the "License");
5// you may not use this file except in compliance with the License.
6// You may obtain a copy of the License at
7//
8//     http://www.apache.org/licenses/LICENSE-2.0
9//
10// Unless required by applicable law or agreed to in writing, software
11// distributed under the License is distributed on an "AS IS" BASIS,
12// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13// See the License for the specific language governing permissions and
14// limitations under the License.
15//
16
17//! Rust port of the Java `ElasticQueue` + `FileElasticStore`
18//! (`org.platformlambda.core.util`) — the **reactive back-pressure overflow
19//! buffer** behind every route's manager (`ServiceQueue` analog).
20//!
21//! A per-route two-tier FIFO: the first [`MEMORY_BUFFER`] events of a burst stay
22//! in memory, then the overflow spills to fixed-size append-only **segment**
23//! files. Record format is portable and kept **byte-identical to the Java file
24//! store**: `[4-byte big-endian length][payload]`. A segment is sealed once it
25//! passes the size threshold (`elastic.queue.segment.size.bytes`, default 16 MB,
26//! min 512); a sealed, fully-consumed segment is **deleted immediately** — O(1)
27//! reclamation, no compaction, no cleaner thread (the whole point vs. the legacy
28//! Berkeley DB store, which is **not ported** — maintainer decision 2026-07-15;
29//! with one store the Java `ElasticStore` strategy facade collapses into this
30//! single type).
31//!
32//! Threading: **single-threaded per route** (driven by the route's manager
33//! task). The buffer is transient — not durable across restart; there is no
34//! fsync. Base directory: `transient.data.store` (default `/tmp/reactive`,
35//! kept verbatim — D9) + `<application name>-<process origin>` (Platform
36//! identity) unless `running.in.cloud=true`.
37//!
38//! **Housekeeping** (Java `ensureBaseDir` + periodic keep-alive + shutdown
39//! hook, wired by the lifecycle): [`start_housekeeping`] refreshes the RUNNING
40//! liveness marker every 20 s and runs [`scan_expired_stores`] once (removes
41//! sibling holding areas whose marker is stale for > 1 h, or unknown dirs
42//! holding segment files); [`shutdown_cleanup`] purges segments + the marker on
43//! graceful exit (Java uses a JVM shutdown hook; signal handling is a later
44//! increment — call it from your main).
45
46use std::collections::VecDeque;
47use std::fs::{File, OpenOptions};
48use std::io::{Read, Seek, SeekFrom, Write};
49use std::path::{Path, PathBuf};
50use std::sync::OnceLock;
51use std::time::Duration;
52
53use crate::function::AppError;
54use crate::platform::Platform;
55use crate::util::app_config_reader::AppConfigReader;
56
57/// First this many events of a burst are held in memory before spilling to disk
58/// (Java `ElasticStore.MEMORY_BUFFER`).
59pub const MEMORY_BUFFER: u64 = 20;
60
61const LENGTH_PREFIX: usize = 4;
62const SEGMENT_PREFIX: &str = "eq-";
63const SEGMENT_SUFFIX: &str = ".dat";
64const DEFAULT_SEGMENT_BYTES: u64 = 16 * 1024 * 1024;
65const MIN_SEGMENT_BYTES: u64 = 512;
66const SEGMENT_SIZE_CONFIG: &str = "elastic.queue.segment.size.bytes";
67const RUNNING: &str = "RUNNING";
68
69static BASE_DIR: OnceLock<PathBuf> = OnceLock::new();
70
71const KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(20);
72const EXPIRY: Duration = Duration::from_secs(60 * 60); // one hour (Java ONE_HOUR)
73
74/// The holding area for segment files, created once per process
75/// (Java `FileElasticStore.ensureBaseDir`).
76pub fn base_dir() -> &'static Path {
77    BASE_DIR.get_or_init(|| {
78        let config = AppConfigReader::get_instance();
79        let tmp_root =
80            PathBuf::from(config.get_property_or("transient.data.store", "/tmp/reactive"));
81        let running_in_cloud = config.get_property_or("running.in.cloud", "false") == "true";
82        let dir = if running_in_cloud {
83            tmp_root
84        } else {
85            tmp_root.join(format!("{}-{}", Platform::name(), Platform::origin()))
86        };
87        if !dir.exists() {
88            if let Err(e) = std::fs::create_dir_all(&dir) {
89                log::error!("Unable to create {} - {e}", dir.display());
90            } else {
91                log::info!("{} created", dir.display());
92            }
93        }
94        write_running_marker(&dir);
95        purge_leftover_segments(&dir, None);
96        log::info!("Elastic file store ready ({})", dir.display());
97        dir
98    })
99}
100
101/// Refresh the RUNNING liveness marker (Java `keepAlive`).
102fn write_running_marker(dir: &Path) {
103    let epoch_ms = std::time::SystemTime::now()
104        .duration_since(std::time::UNIX_EPOCH)
105        .map(|d| d.as_millis().to_string())
106        .unwrap_or_default();
107    let _ = std::fs::write(dir.join(RUNNING), epoch_ms);
108}
109
110/// Start the housekeeping the lifecycle owns (idempotent — Java wires this in
111/// `ensureBaseDir` + a Vert.x periodic timer): refresh the RUNNING marker every
112/// 20 s so siblings can tell this holding area is alive, and scan once for
113/// expired/unknown holding areas. Must be called within a Tokio runtime
114/// (`AppStarter::run` does — its "essential services" phase).
115pub fn start_housekeeping() {
116    static STARTED: OnceLock<()> = OnceLock::new();
117    STARTED.get_or_init(|| {
118        let dir = base_dir().to_path_buf();
119        let running_in_cloud =
120            AppConfigReader::get_instance().get_property_or("running.in.cloud", "false") == "true";
121        if !running_in_cloud {
122            scan_expired_stores();
123        }
124        tokio::spawn(async move {
125            loop {
126                tokio::time::sleep(KEEP_ALIVE_INTERVAL).await;
127                write_running_marker(&dir);
128            }
129        });
130    });
131}
132
133/// Remove sibling holding areas left behind by dead processes (Java
134/// `scanExpiredStores`/`removeExpiredStore`): a dir whose RUNNING marker went
135/// stale (> 1 h) while still holding segment files has expired; a dir holding
136/// segment files with **no** marker is unknown debris. Both are removed.
137pub fn scan_expired_stores() {
138    let current = base_dir();
139    let Some(tmp_root) = current.parent() else {
140        return;
141    };
142    let Ok(entries) = std::fs::read_dir(tmp_root) else {
143        return;
144    };
145    for entry in entries.flatten() {
146        let folder = entry.path();
147        if !folder.is_dir() || folder == current {
148            continue;
149        }
150        let marker = folder.join(RUNNING);
151        if marker.exists() {
152            let stale = std::fs::metadata(&marker)
153                .and_then(|m| m.modified())
154                .ok()
155                .and_then(|modified| modified.elapsed().ok())
156                .is_some_and(|age| age > EXPIRY);
157            if stale && has_segment_files(&folder) && std::fs::remove_dir_all(&folder).is_ok() {
158                log::info!("Elastic file holding area {} expired", folder.display());
159            }
160        } else if has_segment_files(&folder) && std::fs::remove_dir_all(&folder).is_ok() {
161            log::warn!(
162                "Unknown elastic file holding area {} removed",
163                folder.display()
164            );
165        }
166    }
167}
168
169fn has_segment_files(folder: &Path) -> bool {
170    std::fs::read_dir(folder)
171        .map(|entries| {
172            entries.flatten().any(|e| {
173                let name = e.file_name().to_string_lossy().to_string();
174                e.path().is_file()
175                    && name.starts_with(SEGMENT_PREFIX)
176                    && name.ends_with(SEGMENT_SUFFIX)
177            })
178        })
179        .unwrap_or(false)
180}
181
182/// Graceful-exit cleanup (Java's JVM shutdown hook): purge this process's
183/// segment files and remove the RUNNING marker. Call from your main after the
184/// lifecycle completes; OS-signal wiring is a later increment.
185pub fn shutdown_cleanup() {
186    let dir = base_dir();
187    purge_leftover_segments(dir, None);
188    if let Err(e) = std::fs::remove_file(dir.join(RUNNING)) {
189        log::debug!("Unable to delete {RUNNING} marker - {e}");
190    }
191}
192
193/// Best-effort removal of leftover segment files — all of them, or only those
194/// for one sanitized route id (Java `purgeLeftoverSegments`).
195fn purge_leftover_segments(dir: &Path, id_prefix: Option<&str>) {
196    let prefix = match id_prefix {
197        Some(id) => format!("{SEGMENT_PREFIX}{id}-"),
198        None => SEGMENT_PREFIX.to_string(),
199    };
200    if let Ok(entries) = std::fs::read_dir(dir) {
201        for entry in entries.flatten() {
202            let name = entry.file_name().to_string_lossy().to_string();
203            if name.starts_with(&prefix) && name.ends_with(SEGMENT_SUFFIX) {
204                if let Err(e) = std::fs::remove_file(entry.path()) {
205                    log::debug!("Unable to delete leftover {name} - {e}");
206                }
207            }
208        }
209    }
210}
211
212fn sanitize(s: &str) -> String {
213    s.chars()
214        .map(|c| {
215            if c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-') {
216                c
217            } else {
218                '_'
219            }
220        })
221        .collect()
222}
223
224/// One append-only spill file (Java `FileElasticStore.Segment`): sealed when it
225/// passes the size threshold; reclaimed (deleted) as soon as fully consumed.
226struct Segment {
227    index: u32,
228    path: PathBuf,
229    file: Option<File>,
230    write_pos: u64,
231    read_pos: u64,
232    records_written: u64,
233    records_read: u64,
234    sealed: bool,
235}
236
237impl Segment {
238    fn open(path: PathBuf, index: u32) -> Result<Self, AppError> {
239        let file = OpenOptions::new()
240            .create(true)
241            .read(true)
242            .write(true)
243            .truncate(true)
244            .open(&path)
245            .map_err(|e| {
246                AppError::new(
247                    500,
248                    format!("Unable to open elastic segment {}: {e}", path.display()),
249                )
250            })?;
251        Ok(Segment {
252            index,
253            path,
254            file: Some(file),
255            write_pos: 0,
256            read_pos: 0,
257            records_written: 0,
258            records_read: 0,
259            sealed: false,
260        })
261    }
262
263    /// Reopen lazily when a sealed segment becomes the read head (Java parity:
264    /// at most the write tail and the read head hold open files).
265    fn handle(&mut self) -> Result<&mut File, AppError> {
266        if self.file.is_none() {
267            let file = OpenOptions::new()
268                .read(true)
269                .write(true)
270                .open(&self.path)
271                .map_err(|e| {
272                    AppError::new(
273                        500,
274                        format!(
275                            "Unable to reopen elastic segment {}: {e}",
276                            self.path.display()
277                        ),
278                    )
279                })?;
280            self.file = Some(file);
281        }
282        Ok(self.file.as_mut().expect("segment file just ensured"))
283    }
284
285    fn close_quietly(&mut self) {
286        self.file = None; // dropping the handle closes it
287    }
288
289    fn close_and_delete(&mut self) {
290        self.close_quietly();
291        if let Err(e) = std::fs::remove_file(&self.path) {
292            log::debug!("Unable to delete segment {} - {e}", self.path.display());
293        }
294    }
295}
296
297/// The per-route elastic overflow buffer. **Reserved for system use** — the
298/// route manager drives it; do not use directly in application code (Java
299/// carries the same warning on `ServiceQueue`).
300pub struct ElasticQueue {
301    id: String,
302    safe_id: String,
303    segment_bytes: u64,
304    memory: VecDeque<Vec<u8>>,
305    segments: VecDeque<Segment>,
306    read_counter: u64,
307    write_counter: u64,
308    empty: bool,
309    peeked: Option<Vec<u8>>,
310    generation: u32,
311}
312
313impl ElasticQueue {
314    /// `id` is the service route path.
315    pub fn new(id: &str) -> Self {
316        let safe_id = sanitize(id);
317        let segment_bytes = AppConfigReader::get_instance()
318            .get_property_or(SEGMENT_SIZE_CONFIG, &DEFAULT_SEGMENT_BYTES.to_string())
319            .parse::<u64>()
320            .unwrap_or(DEFAULT_SEGMENT_BYTES)
321            .max(MIN_SEGMENT_BYTES);
322        base_dir(); // ensure the holding area exists before any spill
323        ElasticQueue {
324            id: id.to_string(),
325            safe_id,
326            segment_bytes,
327            memory: VecDeque::new(),
328            segments: VecDeque::new(),
329            read_counter: 0,
330            write_counter: 0,
331            empty: true,
332            peeked: None,
333            generation: 0,
334        }
335    }
336
337    pub fn id(&self) -> &str {
338        &self.id
339    }
340
341    pub fn read_counter(&self) -> u64 {
342        self.read_counter
343    }
344
345    pub fn write_counter(&self) -> u64 {
346        self.write_counter
347    }
348
349    /// The queue is closed when the counters are reset (Java `isClosed`).
350    pub fn is_closed(&self) -> bool {
351        self.write_counter == 0
352    }
353
354    /// Append one serialized event: memory tier first, then disk spill
355    /// (Java `write`). Empty payloads are ignored.
356    pub fn write(&mut self, event: &[u8]) -> Result<(), AppError> {
357        if event.is_empty() {
358            return Ok(());
359        }
360        if self.write_counter < MEMORY_BUFFER {
361            // for highest performance, save to memory for the first few blocks
362            self.memory.push_back(event.to_vec());
363        } else {
364            self.append_to_disk(event)?;
365        }
366        self.write_counter += 1;
367        self.empty = false;
368        Ok(())
369    }
370
371    /// Look at the next event without consuming it (Java `peek`).
372    pub fn peek(&mut self) -> Result<Vec<u8>, AppError> {
373        if let Some(held) = &self.peeked {
374            return Ok(held.clone());
375        }
376        let next = self.read()?;
377        if !next.is_empty() {
378            self.peeked = Some(next.clone());
379        }
380        Ok(next)
381    }
382
383    /// Consume the next event in FIFO order; an empty vec means the queue has
384    /// caught up with writes (and the counters were reset — Java `read`).
385    pub fn read(&mut self) -> Result<Vec<u8>, AppError> {
386        if let Some(held) = self.peeked.take() {
387            return Ok(held);
388        }
389        if self.read_counter >= self.write_counter {
390            // catch up with writes and thus nothing to read
391            self.close();
392            return Ok(Vec::new());
393        }
394        if self.read_counter < MEMORY_BUFFER {
395            if let Some(event) = self.memory.pop_front() {
396                self.read_counter += 1;
397                return Ok(event);
398            }
399            return Ok(Vec::new());
400        }
401        let event = self.read_from_disk()?;
402        if !event.is_empty() {
403            self.read_counter += 1;
404        }
405        Ok(event)
406    }
407
408    /// Reset the queue once drained (Java `close` → `resetCounter`).
409    pub fn close(&mut self) {
410        self.peeked = None;
411        if !self.empty {
412            self.empty = true;
413            self.read_counter = 0;
414            self.write_counter = 0;
415            self.memory.clear();
416            for segment in &mut self.segments {
417                segment.close_and_delete();
418            }
419            self.segments.clear();
420            self.generation += 1;
421        }
422    }
423
424    /// Final clean-up when the route leaves service (Java `destroy`): close and
425    /// remove any stray segment files for this route across generations.
426    pub fn destroy(&mut self) {
427        self.close();
428        purge_leftover_segments(base_dir(), Some(&self.safe_id));
429    }
430
431    fn append_to_disk(&mut self, event: &[u8]) -> Result<(), AppError> {
432        let needs_new_tail = match self.segments.back() {
433            None => true,
434            Some(tail) => tail.sealed,
435        };
436        if needs_new_tail {
437            let index = self.segments.back().map_or(0, |t| t.index + 1);
438            let path = base_dir().join(format!(
439                "{SEGMENT_PREFIX}{}-{}-{}{SEGMENT_SUFFIX}",
440                self.safe_id, self.generation, index
441            ));
442            self.segments.push_back(Segment::open(path, index)?);
443        }
444        let single_segment = self.segments.len() == 1;
445        let segment_bytes = self.segment_bytes;
446        let tail = self.segments.back_mut().expect("tail segment just ensured");
447        // portable record format, byte-identical to the Java store:
448        // [4-byte big-endian length][payload]
449        let mut record = Vec::with_capacity(LENGTH_PREFIX + event.len());
450        record.extend_from_slice(&(event.len() as u32).to_be_bytes());
451        record.extend_from_slice(event);
452        let write_pos = tail.write_pos;
453        let file = tail.handle()?;
454        file.seek(SeekFrom::Start(write_pos))
455            .and_then(|_| file.write_all(&record))
456            .map_err(|e| AppError::new(500, format!("Elastic spill write failed - {e}")))?;
457        tail.write_pos += record.len() as u64;
458        tail.records_written += 1;
459        if tail.write_pos >= segment_bytes {
460            tail.sealed = true;
461            // keep only the read head open (the tail is not the head unless
462            // this is the sole segment)
463            if !single_segment {
464                tail.close_quietly();
465            }
466        }
467        Ok(())
468    }
469
470    fn read_from_disk(&mut self) -> Result<Vec<u8>, AppError> {
471        let Some(head) = self.segments.front_mut() else {
472            log::error!(
473                "Missing segment for {} at read position {}",
474                self.id,
475                self.read_counter
476            );
477            return Ok(Vec::new());
478        };
479        let read_pos = head.read_pos;
480        let file = head.handle()?;
481        let mut len_buf = [0u8; LENGTH_PREFIX];
482        file.seek(SeekFrom::Start(read_pos))
483            .and_then(|_| file.read_exact(&mut len_buf))
484            .map_err(|e| AppError::new(500, format!("Elastic spill read failed - {e}")))?;
485        let len = u32::from_be_bytes(len_buf) as usize;
486        let mut payload = vec![0u8; len];
487        file.read_exact(&mut payload)
488            .map_err(|e| AppError::new(500, format!("Elastic spill read failed - {e}")))?;
489        head.read_pos += (LENGTH_PREFIX + len) as u64;
490        head.records_read += 1;
491        // a sealed, fully-consumed segment is reclaimed immediately
492        // (O(1), no cleaner thread)
493        if head.sealed && head.records_read >= head.records_written {
494            head.close_and_delete();
495            self.segments.pop_front();
496        }
497        Ok(payload)
498    }
499}
500
501impl Drop for ElasticQueue {
502    fn drop(&mut self) {
503        self.destroy();
504    }
505}
platform_core/util/elastic_queue.rs

platform_core/util/
elastic_queue.rs
