phemios_core/
concerts_budget.rs

1//! Device-wide conservative admission: at most 1,000 attempted event requests in any 32 days.
2use serde_json::{Value, json};
3use std::{
4    fs::{self, File, OpenOptions},
5    io::{Read, Write},
6    path::{Path, PathBuf},
7    time::{SystemTime, UNIX_EPOCH},
8};
9const WINDOW: u64 = 32 * 86_400;
10const LIMIT: usize = 1000;
11const ERROR: &str = "Budget JamBase illisible ou occupé. Aucune recherche envoyée.";
12pub(super) const EXHAUSTED: &str = "Plafond JamBase atteint. Aucune recherche envoyée.";
13/// Cross-process lease is retained until the HTTP operation finishes, including quota preflight.
14pub struct Budget {
15    _lease: Lease,
16    root: PathBuf,
17    attempts: Vec<u64>,
18    now: u64,
19    remote: Option<Value>,
20}
21// Closing alone can retain a Unix lock through a duplicated/inherited descriptor.
22// Construct only after successful acquisition, including before fallible journal reads.
23struct Lease(File);
24impl Lease {
25    fn acquire(file: File) -> Result<Self, String> {
26        file.try_lock().map_err(|_| ERROR)?;
27        Ok(Self(file))
28    }
29}
30impl Drop for Lease {
31    fn drop(&mut self) {
32        if self.0.unlock().is_err() {
33            // Drop must not panic or disclose a credential, journal path or OS error text.
34            let _ = writeln!(std::io::stderr().lock(), "JamBase budget unlock failed.");
35        }
36    }
37}
38fn admitted(path: &Path) -> Result<bool, String> {
39    let meta = match fs::symlink_metadata(path) {
40        Ok(meta) => meta,
41        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false),
42        Err(_) => return Err(ERROR.into()),
43    };
44    if !meta.is_file() || meta.len() > 64 * 1024 {
45        return Err(ERROR.into());
46    }
47    #[cfg(unix)]
48    {
49        use std::os::unix::fs::PermissionsExt;
50        if meta.permissions().mode() & 0o077 != 0 {
51            return Err(ERROR.into());
52        }
53    }
54    Ok(true)
55}
56fn options() -> OpenOptions {
57    let mut options = OpenOptions::new();
58    options.write(true);
59    #[cfg(unix)]
60    {
61        use std::os::unix::fs::OpenOptionsExt;
62        options.mode(0o600);
63    }
64    options
65}
66impl Budget {
67    /// Lock and read private retained attempts. Missing stores start empty; malformed ones fail closed.
68    pub fn open(root: PathBuf) -> Result<Self, String> {
69        let now = SystemTime::now()
70            .duration_since(UNIX_EPOCH)
71            .map_err(|_| ERROR)?
72            .as_secs();
73        Self::at(root, now)
74    }
75    fn at(root: PathBuf, now: u64) -> Result<Self, String> {
76        crate::private_storage::create_root(&root)?;
77        let lock = root.join("jambase-budget.lock");
78        admitted(&lock)?;
79        let lease = options()
80            .create(true)
81            .truncate(false)
82            .open(lock)
83            .map_err(|_| ERROR)?;
84        let lease = Lease::acquire(lease)?;
85        Self::read_retained(root, now, lease)
86    }
87    fn read_retained(root: PathBuf, now: u64, lease: Lease) -> Result<Self, String> {
88        let path = root.join("jambase-budget.json");
89        let mut attempts = Vec::new();
90        let mut remote = None;
91        if admitted(&path)? {
92            let mut bytes = Vec::new();
93            File::open(path)
94                .map_err(|_| ERROR)?
95                .take(64 * 1024 + 1)
96                .read_to_end(&mut bytes)
97                .map_err(|_| ERROR)?;
98            if bytes.len() > 64 * 1024 {
99                return Err(ERROR.into());
100            }
101            let raw: Value = serde_json::from_slice(&bytes).map_err(|_| ERROR)?;
102            let values = raw["attempts"].as_array().ok_or(ERROR)?;
103            if raw["version"] != 1 || values.len() > LIMIT {
104                return Err(ERROR.into());
105            }
106            for value in values {
107                attempts.push(value.as_u64().ok_or(ERROR)?);
108            }
109            if attempts.windows(2).any(|w| w[0] > w[1])
110                || attempts.last().is_some_and(|last| *last > now)
111            {
112                return Err(ERROR.into());
113            }
114            attempts.retain(|at| now.saturating_sub(*at) < WINDOW);
115            if let Some(value) = raw.get("remote")
116                && !value.is_null()
117            {
118                remote_period(value)?;
119                let ceiling = retained_ceiling(value)?;
120                if value["usedCalls"]
121                    .as_u64()
122                    .is_none_or(|used| used > ceiling)
123                {
124                    return Err(ERROR.into());
125                }
126                remote = Some(value.clone());
127            }
128        }
129        Ok(Self {
130            _lease: lease,
131            root,
132            attempts,
133            now,
134            remote,
135        })
136    }
137    /// A public projection has no key, filesystem path or account identity.
138    pub fn status(&self) -> Value {
139        json!({"limit":LIMIT,"used":self.attempts.len(),"remaining":LIMIT-self.attempts.len(),"resetAt":self.attempts.first().map(|at|at+WINDOW)})
140    }
141    /// Include provider usage without trusting an eventually consistent counter to rise immediately.
142    pub(super) fn reconcile(&mut self, quota: &Value) -> Result<(), String> {
143        let period = remote_period(quota)?;
144        let observed = quota["usedCalls"].as_u64().ok_or(ERROR)?;
145        let previous = self
146            .remote
147            .as_ref()
148            .filter(|value| remote_period(value).ok() == Some(period))
149            .and_then(|value| value["usedCalls"].as_u64())
150            .unwrap_or(0);
151        let used = observed.max(previous);
152        let ceiling = quota["quota"]
153            .as_u64()
154            .ok_or(ERROR)?
155            .min(LIMIT as u64)
156            .min(observed.saturating_add(quota["remainingCalls"].as_u64().ok_or(ERROR)?));
157        if used >= ceiling {
158            return Err(EXHAUSTED.into());
159        }
160        let added = usize::try_from(used.saturating_sub(previous)).map_err(|_| ERROR)?;
161        if added >= LIMIT.saturating_sub(self.attempts.len()) {
162            return Err(EXHAUSTED.into());
163        }
164        self.attempts.extend(std::iter::repeat_n(self.now, added));
165        self.remote = Some(
166            json!({"periodStart":period.0,"periodEnd":period.1,"usedCalls":used,"ceiling":ceiling}),
167        );
168        Ok(())
169    }
170    /// Persist the charge before sending; failures and ambiguous timeouts are never refunded.
171    pub(super) fn reserve(&mut self) -> Result<(), String> {
172        if self.attempts.len() >= LIMIT {
173            return Err(EXHAUSTED.into());
174        }
175        let mut remote = self.remote.clone();
176        if let Some(value) = &mut remote {
177            let used = value["usedCalls"].as_u64().ok_or(ERROR)?;
178            if used >= retained_ceiling(value)? {
179                return Err(EXHAUSTED.into());
180            }
181            value["usedCalls"] = (used + 1).into();
182        }
183        let pending = self.root.join("jambase-budget.pending");
184        let mut file = options()
185            .create_new(true)
186            .open(&pending)
187            .map_err(|_| ERROR)?;
188        let mut attempts = self.attempts.clone();
189        attempts.push(self.now);
190        let result: Result<(), String> = (|| {
191            file.write_all(
192                json!({"version":1,"attempts":attempts,"remote":remote})
193                    .to_string()
194                    .as_bytes(),
195            )
196            .map_err(|_| ERROR)?;
197            file.sync_all().map_err(|_| ERROR)?;
198            fs::rename(&pending, self.root.join("jambase-budget.json")).map_err(|_| ERROR)?;
199            // Sync the containing directory on Unix so a crash cannot roll back a published charge.
200            #[cfg(unix)]
201            File::open(&self.root)
202                .and_then(|f| f.sync_all())
203                .map_err(|_| ERROR)?;
204            Ok(())
205        })();
206        if result.is_err() {
207            let _ = fs::remove_file(pending);
208        }
209        result?;
210        self.attempts = attempts;
211        self.remote = remote;
212        Ok(())
213    }
214}
215// Version-one ledgers written before effective-quota admission used the fixed 1,000 ceiling.
216fn retained_ceiling(value: &Value) -> Result<u64, String> {
217    match value.get("ceiling") {
218        None => Ok(LIMIT as u64),
219        Some(value) => value
220            .as_u64()
221            .filter(|ceiling| *ceiling <= LIMIT as u64)
222            .ok_or_else(|| ERROR.into()),
223    }
224}
225
226fn remote_period(value: &Value) -> Result<(&str, &str), String> {
227    let field = |name| {
228        value[name].as_str().filter(|text| {
229            (20..=40).contains(&text.len())
230                && text.is_ascii()
231                && text.contains('T')
232                && text.ends_with('Z')
233        })
234    };
235    let start = field("periodStart").ok_or(ERROR)?;
236    let end = field("periodEnd").ok_or(ERROR)?;
237    if start >= end {
238        return Err(ERROR.into());
239    }
240    Ok((start, end))
241}
242#[cfg(test)]
243#[path = "concerts_budget_tests.rs"]
244mod tests;
phemios_core/concerts_budget.rs

phemios_core/
concerts_budget.rs
