Struct Service 

pub struct Service {
    store: Store,
    broker: Arc<Broker>,
    config: Arc<ConfigState>,
    permits: Arc<Semaphore>,
    memory: Arc<Semaphore>,
    batches: Mutex<HashMap<String, Arc<AtomicBool>>>,
    inflight: Arc<Mutex<HashMap<String, usize>>>,
    session_budget: Mutex<()>,
}

Fields

store: Store

broker: Arc<Broker>

config: Arc<ConfigState>

permits: Arc<Semaphore>

Bounds concurrent jobs (CPU).

memory: Arc<Semaphore>

Bounds the decode memory of concurrent jobs, in MiB.

batches: Mutex<HashMap<String, Arc<AtomicBool>>>

Cancellation flags of batches still running.

inflight: Arc<Mutex<HashMap<String, usize>>>

Jobs currently processing each id. Identical images processed at once share one stored original, so a failing job may only discard it when no other job is still using it. Placing an original and the discard check both happen under this lock.

session_budget: Mutex<()>

Serializes the per-session inline-budget read, spend and write, so two overlapping turns of one Claude session can’t both spend the same remaining budget or overwrite each other’s count.

Implementations

impl Service

pub fn new(store: Store, broker: Arc<Broker>, config: Arc<ConfigState>) -> Self

pub fn store(&self) -> &Store

pub fn limits(&self) -> Limits

fn start_sweeper(self: &Arc<Self>)

pub fn ingest( self: &Arc<Self>, req: CommandAttachmentsIngestData, ) -> AttachmentsIngestResult

Classify paths, apply the limits, and (unless dry_run) start processing the accepted images in the background. Returns at once.

pub fn cancel(&self, batch_id: &str) -> bool

pub async fn info(self: &Arc<Self>, ids: &[String]) -> Vec<AttachmentInfo>

Info for each id still in the store, in request order. May decode (after a send-edge change), so call it off the async workers.

pub async fn ensure_derived(self: &Arc<Self>, id: &str) -> Option<StoredMeta>

Metadata under the current fingerprint. When the send edge or the pipeline changed since the attachment was processed, re-derive it from the stored original under the same CPU and memory limits as ingest, so a transcript full of old thumbnails can’t start an unbounded number of decodes at once.

async fn derive_kind( self: &Arc<Self>, id: &str, blob: PathBuf, kind: FileKind, name: &str, edge: u32, ) -> Result<StoredMeta, ProcessError>

Derive a placed original. Images reserve decode memory from their header first; other files are bounded by the extraction limits (extract.rs) and PDFs run out of process. The caller holds a CPU slot.

pub fn with_session_inline_budget<T>( &self, key: &str, carry_from: Option<&str>, max: u64, f: impl FnOnce(u64) -> (T, u64), ) -> T

Run f with the inline budget left in Claude session key (out of max), and record what it reports spending, as one step under a lock. carry_from is a per-block key counted before the session had an id; its count moves into key first. Blocking.

pub fn send_target(&self, id: &str, name: &str) -> Option<SendTarget>

What the agent gets for attachment id sent under the user’s name, marking it used: an image’s send-copy, or for any other file a freshly made copy named name, with a note for the agent’s list. Doesn’t decode: call Self::ensure_derived first so a stale fingerprint is re-derived under the limits.

async fn reserve_memory(&self, w: u32, h: u32) -> Option<OwnedSemaphorePermit>

Reserve decode memory for a w×h image. Held until dropped.

pub async fn commit_upload( self: &Arc<Self>, tmp: PathBuf, id: String, name: String, ) -> Result<AttachmentInfo, ProcessError>

Store a file already sitting in incoming/ (an upload): place it, reserve decode memory, derive. Returns its info.

fn place_tracked( &self, tmp: &Path, id: &str, name: &str, ) -> Result<(PathBuf, FileKind, InflightGuard), ProcessError>

Store::place, registering this job as a user of the id first, under the same lock the discard check takes. Blocking.

async fn discard_unprocessed(&self, guard: InflightGuard)

After a failed job: discard the original unless another job for the same bytes is still processing it (Store::discard_if_unprocessed also keeps it when any fingerprint has metadata). Consumes the guard.

async fn run_batch( self: Arc<Self>, block_id: String, batch_id: String, items: Vec<AttachmentPending>, limits: Limits, cancel: Arc<AtomicBool>, )

async fn run_item( self: &Arc<Self>, block_id: &str, batch_id: &str, item: AttachmentPending, limits: Limits, cancel: &Arc<AtomicBool>, ) -> Result<AttachmentInfo, ProcessError>

One item: wait for a CPU slot, copy + hash + place, reserve decode memory from the header, derive. The cancel flag is checked before each step and between copy chunks; a decode already running finishes but is reported as cancelled.

async fn derive_placed( self: &Arc<Self>, block_id: &str, batch_id: &str, item: &AttachmentPending, id: &str, blob: PathBuf, kind: FileKind, limits: Limits, cancel: &Arc<AtomicBool>, ) -> Result<AttachmentInfo, ProcessError>

The part of Self::run_item after the original is in blobs/: reserve decode memory from the header and derive.

fn copy_and_place( &self, block_id: &str, batch_id: &str, item: &AttachmentPending, limits: Limits, cancel: &AtomicBool, ) -> Result<(String, PathBuf, FileKind, InflightGuard), ProcessError>

Copy the source into incoming/ with progress and cancellation, then move it into blobs/. Runs on a blocking thread.

fn publish<E: Event>(&self, block_id: &str, data: &E::Payload)