a3s_code_core/memory/
maintenance.rs

1//! Explicitly owned maintenance for long-lived memory work.
2//!
3//! Storage construction is intentionally side-effect free. A host starts this
4//! runtime, observes its health, and closes it at the same lifecycle boundary
5//! that owns the associated [`AgentMemory`]. Verified semantic index refresh is
6//! available as an opt-in built-in schedule; consolidation remains a host policy
7//! supplied through [`MemoryMaintenanceJob`].
8
9use super::semantic_refresh::ScheduledSemanticRefreshClaim;
10use super::{AgentMemory, ScheduledSemanticRefresh, SEMANTIC_REFRESH_JOB_NAME};
11use async_trait::async_trait;
12use futures::FutureExt;
13use serde::{Deserialize, Serialize};
14use std::collections::HashSet;
15use std::panic::AssertUnwindSafe;
16use std::sync::{Arc, Mutex, RwLock};
17use std::time::Duration;
18use tokio::task::JoinHandle;
19use tokio_util::sync::CancellationToken;
20
21const PRUNE_JOB_NAME: &str = "v1_prune";
22const DEFAULT_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5);
23const MAX_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(30);
24const MIN_JOB_INTERVAL: Duration = Duration::from_secs(1);
25const MAX_JOB_INTERVAL: Duration = Duration::from_secs(365 * 24 * 60 * 60);
26const MAX_JOB_NAME_BYTES: usize = 64;
27const MAX_OWNER_ID_BYTES: usize = 256;
28const MAX_JOBS: usize = 32;
29const MAX_ERROR_BYTES: usize = 1_024;
30
31/// One policy-owned memory maintenance operation.
32///
33/// Implementations may perform verified consolidation, retention projection,
34/// or other bounded host work. The storage kernel never constructs one. Runs
35/// for the same scheduled job are serialized, and `cancellation` fires when
36/// the owning runtime closes.
37#[async_trait]
38pub trait MemoryMaintenanceJob: Send + Sync {
39    async fn run(
40        &self,
41        context: &MemoryMaintenanceContext,
42        cancellation: CancellationToken,
43    ) -> anyhow::Result<MemoryMaintenanceOutcome>;
44}
45
46/// Exact runtime context passed to a host-owned maintenance job.
47#[derive(Clone)]
48pub struct MemoryMaintenanceContext {
49    owner_id: Arc<str>,
50    memory: Arc<AgentMemory>,
51}
52
53impl MemoryMaintenanceContext {
54    pub fn owner_id(&self) -> &str {
55        &self.owner_id
56    }
57
58    pub fn memory(&self) -> &Arc<AgentMemory> {
59        &self.memory
60    }
61
62    /// Return the exact V2 binding supplied to this memory instance, if any.
63    pub fn durable_memory(&self) -> Option<&crate::durable_memory::DurableMemorySession> {
64        self.memory.durable_memory()
65    }
66}
67
68impl std::fmt::Debug for MemoryMaintenanceContext {
69    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
70        formatter
71            .debug_struct("MemoryMaintenanceContext")
72            .field("owner_id", &self.owner_id)
73            .field("durable_memory", &self.durable_memory())
74            .finish_non_exhaustive()
75    }
76}
77
78/// Machine-readable result of one maintenance run.
79#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
80#[serde(rename_all = "camelCase")]
81pub struct MemoryMaintenanceOutcome {
82    pub affected_items: usize,
83}
84
85impl MemoryMaintenanceOutcome {
86    pub const fn new(affected_items: usize) -> Self {
87        Self { affected_items }
88    }
89}
90
91/// A typed job plus its non-overlapping periodic schedule.
92#[derive(Clone)]
93#[must_use = "a scheduled job does nothing until installed in maintenance options"]
94pub struct ScheduledMemoryMaintenance {
95    name: String,
96    interval: Duration,
97    job: Arc<dyn MemoryMaintenanceJob>,
98}
99
100impl ScheduledMemoryMaintenance {
101    pub fn try_new(
102        name: impl Into<String>,
103        interval: Duration,
104        job: Arc<dyn MemoryMaintenanceJob>,
105    ) -> Result<Self, MemoryMaintenanceError> {
106        let name = name.into();
107        validate_job_name(&name)?;
108        validate_interval(interval)?;
109        Ok(Self {
110            name,
111            interval,
112            job,
113        })
114    }
115
116    pub fn name(&self) -> &str {
117        &self.name
118    }
119
120    pub fn interval(&self) -> Duration {
121        self.interval
122    }
123}
124
125impl std::fmt::Debug for ScheduledMemoryMaintenance {
126    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
127        formatter
128            .debug_struct("ScheduledMemoryMaintenance")
129            .field("name", &self.name)
130            .field("interval", &self.interval)
131            .field("job", &"<host-injected>")
132            .finish()
133    }
134}
135
136/// Typed, session-owned memory maintenance schedules and shutdown policy.
137#[derive(Clone, Debug)]
138#[must_use = "maintenance options do nothing until installed on a session or runtime"]
139pub struct MemoryMaintenanceOptions {
140    jobs: Vec<ScheduledMemoryMaintenance>,
141    semantic_refresh: Option<ScheduledSemanticRefresh>,
142    shutdown_timeout: Duration,
143}
144
145impl Default for MemoryMaintenanceOptions {
146    fn default() -> Self {
147        Self {
148            jobs: Vec::new(),
149            semantic_refresh: None,
150            shutdown_timeout: DEFAULT_SHUTDOWN_TIMEOUT,
151        }
152    }
153}
154
155impl MemoryMaintenanceOptions {
156    pub fn new() -> Self {
157        Self::default()
158    }
159
160    pub fn with_job(mut self, job: ScheduledMemoryMaintenance) -> Self {
161        self.jobs.push(job);
162        self
163    }
164
165    /// Install or replace the single built-in verified semantic refresh
166    /// schedule.
167    pub fn with_semantic_refresh(mut self, schedule: ScheduledSemanticRefresh) -> Self {
168        self.semantic_refresh = Some(schedule);
169        self
170    }
171
172    pub fn try_with_shutdown_timeout(
173        mut self,
174        timeout: Duration,
175    ) -> Result<Self, MemoryMaintenanceError> {
176        validate_shutdown_timeout(timeout)?;
177        self.shutdown_timeout = timeout;
178        Ok(self)
179    }
180
181    pub fn jobs(&self) -> &[ScheduledMemoryMaintenance] {
182        &self.jobs
183    }
184
185    pub fn semantic_refresh(&self) -> Option<&ScheduledSemanticRefresh> {
186        self.semantic_refresh.as_ref()
187    }
188
189    pub fn shutdown_timeout(&self) -> Duration {
190        self.shutdown_timeout
191    }
192
193    pub(crate) fn is_empty(&self) -> bool {
194        self.jobs.is_empty() && self.semantic_refresh.is_none()
195    }
196}
197
198/// Lifecycle state of one owned maintenance runtime.
199#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
200#[serde(rename_all = "snake_case")]
201pub enum MemoryMaintenancePhase {
202    Disabled,
203    Running,
204    Degraded,
205    Closing,
206    Closed,
207}
208
209/// Current health of one scheduled job.
210#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
211#[serde(rename_all = "camelCase")]
212pub struct MemoryMaintenanceJobHealth {
213    pub name: String,
214    pub interval_ms: u64,
215    pub worker_alive: bool,
216    pub run_in_progress: bool,
217    pub successful_runs: u64,
218    pub failed_runs: u64,
219    pub total_affected_items: u64,
220    pub last_affected_items: Option<usize>,
221    pub last_error: Option<String>,
222}
223
224impl MemoryMaintenanceJobHealth {
225    fn new(schedule: &ScheduledMemoryMaintenance) -> Self {
226        Self {
227            name: schedule.name.clone(),
228            interval_ms: duration_ms(schedule.interval),
229            worker_alive: true,
230            run_in_progress: false,
231            successful_runs: 0,
232            failed_runs: 0,
233            total_affected_items: 0,
234            last_affected_items: None,
235            last_error: None,
236        }
237    }
238}
239
240/// Non-sensitive snapshot for readiness and operational diagnostics.
241#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
242#[serde(rename_all = "camelCase")]
243pub struct MemoryMaintenanceHealth {
244    pub phase: MemoryMaintenancePhase,
245    pub jobs: Vec<MemoryMaintenanceJobHealth>,
246}
247
248impl MemoryMaintenanceHealth {
249    pub fn disabled() -> Self {
250        Self {
251            phase: MemoryMaintenancePhase::Disabled,
252            jobs: Vec::new(),
253        }
254    }
255
256    pub fn is_healthy(&self) -> bool {
257        matches!(
258            self.phase,
259            MemoryMaintenancePhase::Disabled
260                | MemoryMaintenancePhase::Running
261                | MemoryMaintenancePhase::Closed
262        )
263    }
264}
265
266/// Bounded shutdown evidence for one maintenance runtime.
267#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)]
268#[serde(rename_all = "camelCase")]
269pub struct MemoryMaintenanceCloseReport {
270    pub jobs_joined: usize,
271    pub jobs_aborted: usize,
272    pub join_failures: usize,
273}
274
275impl MemoryMaintenanceCloseReport {
276    pub fn is_clean(&self) -> bool {
277        self.jobs_aborted == 0 && self.join_failures == 0
278    }
279}
280
281/// Invalid configuration or ownership failure when starting maintenance.
282#[derive(Debug, thiserror::Error)]
283#[non_exhaustive]
284pub enum MemoryMaintenanceError {
285    #[error("invalid memory maintenance configuration for {field}: {reason}")]
286    InvalidConfiguration { field: &'static str, reason: String },
287    #[error("memory maintenance requires a Tokio runtime")]
288    AsyncRuntimeRequired,
289    #[error("this AgentMemory already has an active maintenance owner")]
290    AlreadyOwned,
291    #[error("this semantic refresh schedule already has an active maintenance owner")]
292    SemanticRefreshAlreadyOwned,
293    #[error("no memory maintenance jobs are configured")]
294    NoJobsConfigured,
295}
296
297/// Owner of every periodic task associated with one [`AgentMemory`].
298#[must_use = "the owner must be retained and closed to govern its maintenance tasks"]
299pub struct MemoryMaintenanceRuntime {
300    memory: Arc<AgentMemory>,
301    semantic_refresh_claim: Mutex<Option<ScheduledSemanticRefreshClaim>>,
302    lifetime: CancellationToken,
303    health: Arc<RwLock<MemoryMaintenanceHealth>>,
304    tasks: Mutex<Option<Vec<JoinHandle<()>>>>,
305    close_gate: tokio::sync::Mutex<()>,
306    close_report: Mutex<Option<MemoryMaintenanceCloseReport>>,
307    shutdown_timeout: Duration,
308    claim_released: std::sync::atomic::AtomicBool,
309}
310
311impl std::fmt::Debug for MemoryMaintenanceRuntime {
312    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
313        formatter
314            .debug_struct("MemoryMaintenanceRuntime")
315            .field("health", &self.health())
316            .field("shutdown_timeout", &self.shutdown_timeout)
317            .finish_non_exhaustive()
318    }
319}
320
321impl MemoryMaintenanceRuntime {
322    /// Start an explicitly owned runtime. No task is started by
323    /// [`AgentMemory`] construction itself.
324    pub fn start(
325        owner_id: impl Into<String>,
326        memory: Arc<AgentMemory>,
327        options: MemoryMaintenanceOptions,
328    ) -> Result<Arc<Self>, MemoryMaintenanceError> {
329        let owner_id = owner_id.into();
330        let schedules = Self::validated_schedules(&owner_id, &memory, &options)?;
331        let handle = tokio::runtime::Handle::try_current()
332            .map_err(|_| MemoryMaintenanceError::AsyncRuntimeRequired)?;
333        if memory
334            .maintenance_claimed
335            .compare_exchange(
336                false,
337                true,
338                std::sync::atomic::Ordering::AcqRel,
339                std::sync::atomic::Ordering::Acquire,
340            )
341            .is_err()
342        {
343            return Err(MemoryMaintenanceError::AlreadyOwned);
344        }
345        let semantic_refresh_claim = match &options.semantic_refresh {
346            Some(schedule) => match schedule.try_claim() {
347                Ok(claim) => Some(claim),
348                Err(error) => {
349                    memory
350                        .maintenance_claimed
351                        .store(false, std::sync::atomic::Ordering::Release);
352                    return Err(error);
353                }
354            },
355            None => None,
356        };
357
358        let context = MemoryMaintenanceContext {
359            owner_id: Arc::from(owner_id),
360            memory: Arc::clone(&memory),
361        };
362        let health = Arc::new(RwLock::new(MemoryMaintenanceHealth {
363            phase: MemoryMaintenancePhase::Running,
364            jobs: schedules
365                .iter()
366                .map(MemoryMaintenanceJobHealth::new)
367                .collect(),
368        }));
369        let lifetime = CancellationToken::new();
370        let worker_semantic_refresh_claim = semantic_refresh_claim.clone();
371        let runtime = Arc::new(Self {
372            memory,
373            semantic_refresh_claim: Mutex::new(semantic_refresh_claim),
374            lifetime: lifetime.clone(),
375            health: Arc::clone(&health),
376            tasks: Mutex::new(None),
377            close_gate: tokio::sync::Mutex::new(()),
378            close_report: Mutex::new(None),
379            shutdown_timeout: options.shutdown_timeout,
380            claim_released: std::sync::atomic::AtomicBool::new(false),
381        });
382        let tasks = schedules
383            .into_iter()
384            .enumerate()
385            .map(|(index, schedule)| {
386                let context = context.clone();
387                let health = Arc::clone(&health);
388                let cancellation = lifetime.child_token();
389                let semantic_refresh_claim = worker_semantic_refresh_claim.clone();
390                handle.spawn(async move {
391                    let observe_cancellation = cancellation.clone();
392                    let result = AssertUnwindSafe(run_schedule(
393                        schedule,
394                        index,
395                        context,
396                        Arc::clone(&health),
397                        cancellation,
398                    ))
399                    .catch_unwind()
400                    .await;
401                    let mut snapshot = write_unpoisoned(&health);
402                    let job = &mut snapshot.jobs[index];
403                    job.run_in_progress = false;
404                    job.worker_alive = false;
405                    if result.is_err() && !observe_cancellation.is_cancelled() {
406                        job.failed_runs = job.failed_runs.saturating_add(1);
407                        job.last_error = Some("maintenance worker panicked".to_string());
408                    }
409                    drop(semantic_refresh_claim);
410                })
411            })
412            .collect();
413        *lock_unpoisoned(&runtime.tasks) = Some(tasks);
414        Ok(runtime)
415    }
416
417    pub(crate) fn validate_configuration(
418        owner_id: &str,
419        memory: &AgentMemory,
420        options: &MemoryMaintenanceOptions,
421    ) -> Result<bool, MemoryMaintenanceError> {
422        match Self::validated_schedules(owner_id, memory, options) {
423            Ok(_) => Ok(true),
424            Err(MemoryMaintenanceError::NoJobsConfigured) => Ok(false),
425            Err(error) => Err(error),
426        }
427    }
428
429    fn validated_schedules(
430        owner_id: &str,
431        memory: &AgentMemory,
432        options: &MemoryMaintenanceOptions,
433    ) -> Result<Vec<ScheduledMemoryMaintenance>, MemoryMaintenanceError> {
434        validate_owner_id(owner_id)?;
435        validate_shutdown_timeout(options.shutdown_timeout)?;
436        let mut schedules = Vec::new();
437        if let Some((policy, interval)) = memory.maintenance_prune_schedule() {
438            schedules.push(ScheduledMemoryMaintenance::try_new(
439                PRUNE_JOB_NAME,
440                interval,
441                Arc::new(PruneMemoryJob {
442                    store: Arc::clone(memory.store()),
443                    policy,
444                }),
445            )?);
446        }
447        if let Some(semantic_refresh) = &options.semantic_refresh {
448            semantic_refresh.validate_for(memory)?;
449            schedules.push(semantic_refresh.as_maintenance()?);
450        }
451        for schedule in &options.jobs {
452            if matches!(
453                schedule.name.as_str(),
454                PRUNE_JOB_NAME | SEMANTIC_REFRESH_JOB_NAME
455            ) {
456                return Err(invalid(
457                    "jobs.name",
458                    format!("'{}' is reserved for built-in maintenance", schedule.name),
459                ));
460            }
461            schedules.push(schedule.clone());
462        }
463        if schedules.is_empty() {
464            return Err(MemoryMaintenanceError::NoJobsConfigured);
465        }
466        if schedules.len() > MAX_JOBS {
467            return Err(invalid(
468                "jobs",
469                format!("must not contain more than {MAX_JOBS} jobs"),
470            ));
471        }
472        let mut names = HashSet::with_capacity(schedules.len());
473        for schedule in &schedules {
474            validate_job_name(&schedule.name)?;
475            validate_interval(schedule.interval)?;
476            if !names.insert(schedule.name.clone()) {
477                return Err(invalid(
478                    "jobs.name",
479                    format!("duplicate job name '{}'", schedule.name),
480                ));
481            }
482        }
483        Ok(schedules)
484    }
485
486    pub fn health(&self) -> MemoryMaintenanceHealth {
487        let mut health = read_unpoisoned(&self.health).clone();
488        if health.phase == MemoryMaintenancePhase::Running
489            && health
490                .jobs
491                .iter()
492                .any(|job| !job.worker_alive || job.last_error.is_some())
493        {
494            health.phase = MemoryMaintenancePhase::Degraded;
495        }
496        health
497    }
498
499    /// Cancel all jobs, join them within one total deadline, and abort any
500    /// worker that exceeds it. Repeated calls return the first close report.
501    pub async fn close(&self) -> MemoryMaintenanceCloseReport {
502        let _close = self.close_gate.lock().await;
503        if let Some(report) = lock_unpoisoned(&self.close_report).clone() {
504            return report;
505        }
506        write_unpoisoned(&self.health).phase = MemoryMaintenancePhase::Closing;
507        self.lifetime.cancel();
508        let tasks = lock_unpoisoned(&self.tasks).take().unwrap_or_default();
509        let deadline = tokio::time::Instant::now() + self.shutdown_timeout;
510        let mut report = MemoryMaintenanceCloseReport::default();
511        for mut task in tasks {
512            match tokio::time::timeout_at(deadline, &mut task).await {
513                Ok(Ok(())) => report.jobs_joined += 1,
514                Ok(Err(_)) => report.join_failures += 1,
515                Err(_) => {
516                    task.abort();
517                    let _ = task.await;
518                    report.jobs_aborted += 1;
519                }
520            }
521        }
522        {
523            let mut health = write_unpoisoned(&self.health);
524            health.phase = MemoryMaintenancePhase::Closed;
525            for job in &mut health.jobs {
526                job.worker_alive = false;
527                job.run_in_progress = false;
528            }
529        }
530        self.release_claim();
531        *lock_unpoisoned(&self.close_report) = Some(report.clone());
532        report
533    }
534
535    fn release_claim(&self) {
536        if self
537            .claim_released
538            .compare_exchange(
539                false,
540                true,
541                std::sync::atomic::Ordering::AcqRel,
542                std::sync::atomic::Ordering::Acquire,
543            )
544            .is_ok()
545        {
546            self.memory
547                .maintenance_claimed
548                .store(false, std::sync::atomic::Ordering::Release);
549            drop(lock_unpoisoned(&self.semantic_refresh_claim).take());
550        }
551    }
552}
553
554impl Drop for MemoryMaintenanceRuntime {
555    fn drop(&mut self) {
556        self.lifetime.cancel();
557        if let Some(tasks) = lock_unpoisoned(&self.tasks).take() {
558            for task in tasks {
559                task.abort();
560            }
561        }
562        self.release_claim();
563        let mut health = write_unpoisoned(&self.health);
564        health.phase = MemoryMaintenancePhase::Closed;
565        for job in &mut health.jobs {
566            job.worker_alive = false;
567            job.run_in_progress = false;
568        }
569    }
570}
571
572struct PruneMemoryJob {
573    store: Arc<dyn a3s_memory::MemoryStore>,
574    policy: a3s_memory::PrunePolicy,
575}
576
577#[async_trait]
578impl MemoryMaintenanceJob for PruneMemoryJob {
579    async fn run(
580        &self,
581        _context: &MemoryMaintenanceContext,
582        _cancellation: CancellationToken,
583    ) -> anyhow::Result<MemoryMaintenanceOutcome> {
584        self.store
585            .prune(&self.policy)
586            .await
587            .map(MemoryMaintenanceOutcome::new)
588    }
589}
590
591async fn run_schedule(
592    schedule: ScheduledMemoryMaintenance,
593    health_index: usize,
594    context: MemoryMaintenanceContext,
595    health: Arc<RwLock<MemoryMaintenanceHealth>>,
596    cancellation: CancellationToken,
597) {
598    let first_tick = tokio::time::Instant::now() + schedule.interval;
599    let mut ticker = tokio::time::interval_at(first_tick, schedule.interval);
600    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
601    loop {
602        tokio::select! {
603            biased;
604            _ = cancellation.cancelled() => break,
605            _ = ticker.tick() => {}
606        }
607        {
608            let mut snapshot = write_unpoisoned(&health);
609            snapshot.jobs[health_index].run_in_progress = true;
610        }
611        let run_cancellation = cancellation.child_token();
612        let run = schedule.job.run(&context, run_cancellation.clone());
613        tokio::pin!(run);
614        let (result, close_after_run) = tokio::select! {
615            biased;
616            _ = cancellation.cancelled() => {
617                run_cancellation.cancel();
618                (run.await, true)
619            }
620            result = &mut run => (result, false),
621        };
622        let mut snapshot = write_unpoisoned(&health);
623        let job = &mut snapshot.jobs[health_index];
624        job.run_in_progress = false;
625        match result {
626            Ok(outcome) => {
627                job.successful_runs = job.successful_runs.saturating_add(1);
628                job.total_affected_items = job
629                    .total_affected_items
630                    .saturating_add(u64::try_from(outcome.affected_items).unwrap_or(u64::MAX));
631                job.last_affected_items = Some(outcome.affected_items);
632                job.last_error = None;
633            }
634            Err(error) => {
635                let message = bounded_error(error.to_string());
636                job.failed_runs = job.failed_runs.saturating_add(1);
637                job.last_error = Some(message.clone());
638                tracing::warn!(
639                    owner_id = %context.owner_id,
640                    job = %schedule.name,
641                    error = %message,
642                    "Memory maintenance job failed"
643                );
644            }
645        }
646        drop(snapshot);
647        if close_after_run {
648            break;
649        }
650    }
651}
652
653fn validate_job_name(name: &str) -> Result<(), MemoryMaintenanceError> {
654    if name.is_empty() || name.trim() != name {
655        return Err(invalid(
656            "job.name",
657            "must not be empty or contain surrounding whitespace",
658        ));
659    }
660    if name.len() > MAX_JOB_NAME_BYTES {
661        return Err(invalid(
662            "job.name",
663            format!("must not exceed {MAX_JOB_NAME_BYTES} bytes"),
664        ));
665    }
666    if !name
667        .chars()
668        .all(|character| character.is_ascii_alphanumeric() || "_.-".contains(character))
669    {
670        return Err(invalid(
671            "job.name",
672            "must contain only ASCII letters, digits, '.', '_' or '-'",
673        ));
674    }
675    Ok(())
676}
677
678pub(super) fn validate_interval(interval: Duration) -> Result<(), MemoryMaintenanceError> {
679    if interval < MIN_JOB_INTERVAL || interval > MAX_JOB_INTERVAL {
680        return Err(invalid(
681            "job.interval",
682            "must be between one second and 365 days",
683        ));
684    }
685    Ok(())
686}
687
688fn validate_shutdown_timeout(timeout: Duration) -> Result<(), MemoryMaintenanceError> {
689    if timeout.is_zero() || timeout > MAX_SHUTDOWN_TIMEOUT {
690        return Err(invalid(
691            "shutdownTimeout",
692            "must be greater than zero and no longer than 30 seconds",
693        ));
694    }
695    Ok(())
696}
697
698fn validate_owner_id(owner_id: &str) -> Result<(), MemoryMaintenanceError> {
699    if owner_id.trim().is_empty() {
700        return Err(invalid("ownerId", "must not be empty or whitespace"));
701    }
702    if owner_id.len() > MAX_OWNER_ID_BYTES {
703        return Err(invalid(
704            "ownerId",
705            format!("must not exceed {MAX_OWNER_ID_BYTES} bytes"),
706        ));
707    }
708    Ok(())
709}
710
711fn invalid(field: &'static str, reason: impl Into<String>) -> MemoryMaintenanceError {
712    MemoryMaintenanceError::InvalidConfiguration {
713        field,
714        reason: reason.into(),
715    }
716}
717
718fn duration_ms(duration: Duration) -> u64 {
719    duration.as_millis().min(u128::from(u64::MAX)) as u64
720}
721
722fn bounded_error(mut message: String) -> String {
723    if message.len() <= MAX_ERROR_BYTES {
724        return message;
725    }
726    let mut boundary = MAX_ERROR_BYTES;
727    while !message.is_char_boundary(boundary) {
728        boundary -= 1;
729    }
730    message.truncate(boundary);
731    message
732}
733
734fn lock_unpoisoned<T>(lock: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
735    lock.lock()
736        .unwrap_or_else(std::sync::PoisonError::into_inner)
737}
738
739fn read_unpoisoned<T>(lock: &RwLock<T>) -> std::sync::RwLockReadGuard<'_, T> {
740    lock.read()
741        .unwrap_or_else(std::sync::PoisonError::into_inner)
742}
743
744fn write_unpoisoned<T>(lock: &RwLock<T>) -> std::sync::RwLockWriteGuard<'_, T> {
745    lock.write()
746        .unwrap_or_else(std::sync::PoisonError::into_inner)
747}
a3s_code_core/memory/maintenance.rs

a3s_code_core/memory/
maintenance.rs
