qualia_core_db/medical/
dicom_ingest.rs

1//! Core 3 DICOM split-ingestion pipeline.
2//!
3//! Pixel payloads stream to a memory-mapped blob store; semantic metadata becomes
4//! 48-byte Quins appended to the conduct WAL. Core 1 never blocks on I/O — jobs are
5//! handed to a dedicated Swarm worker via a lock-free channel.
6
7use crate::dicom::{
8    decode_blob_pointer, encode_blob_pointer, pack_volume_metadata, split_dicom_payload,
9    DicomMetadata, DicomSplitPayload,
10};
11use crate::q_hash;
12use crate::wal::WriteAheadLog;
13use crate::NQuin;
14use crossbeam_channel::{bounded, Receiver, Sender};
15use memmap2::Mmap;
16use std::cell::UnsafeCell;
17use std::fs::{File, OpenOptions};
18use std::io::{Read, Seek, SeekFrom, Write};
19use std::path::{Path, PathBuf};
20use std::sync::atomic::{AtomicU32, AtomicU64, AtomicU8, AtomicUsize, Ordering};
21use std::sync::OnceLock;
22use std::thread;
23
24pub const BLOB_FILE_NAME: &str = "dicom_pixels.blob";
25pub const MAX_SERIES_RECORDS: usize = 64;
26pub const JOB_PENDING: u8 = 0;
27pub const JOB_COMPLETE: u8 = 1;
28pub const JOB_FAILED: u8 = 2;
29
30/// Published series record — single-writer (Core 3), lock-free readers (Core 1).
31#[derive(Debug, Clone, Copy, Default)]
32pub struct DicomSeriesRecord {
33    pub patient_did_hash: u64,
34    pub series_hash: u64,
35    pub blob_offset: u64,
36    pub blob_length: u32,
37    pub rows: u16,
38    pub cols: u16,
39    pub organ_hash: u64,
40    pub pixel_pointer_quin_parity: u64,
41}
42
43struct IngestJob {
44    job_id: u64,
45    source_path: PathBuf,
46    patient_did_hash: u64,
47    storage_root: PathBuf,
48}
49
50struct JobSlot {
51    status: AtomicU8,
52    series_hash: AtomicU64,
53    blob_offset: AtomicU64,
54    blob_length: AtomicU32,
55}
56
57impl JobSlot {
58    const fn new() -> Self {
59        Self {
60            status: AtomicU8::new(JOB_PENDING),
61            series_hash: AtomicU64::new(0),
62            blob_offset: AtomicU64::new(0),
63            blob_length: AtomicU32::new(0),
64        }
65    }
66}
67
68static IO_TX: OnceLock<Sender<IngestJob>> = OnceLock::new();
69static NEXT_JOB_ID: AtomicU64 = AtomicU64::new(1);
70static JOB_SLOTS: [JobSlot; MAX_SERIES_RECORDS] = [const { JobSlot::new() }; MAX_SERIES_RECORDS];
71const EMPTY_SERIES_RECORD: DicomSeriesRecord = DicomSeriesRecord {
72    patient_did_hash: 0,
73    series_hash: 0,
74    blob_offset: 0,
75    blob_length: 0,
76    rows: 0,
77    cols: 0,
78    organ_hash: 0,
79    pixel_pointer_quin_parity: 0,
80};
81struct SyncSeriesRegistry([UnsafeCell<DicomSeriesRecord>; MAX_SERIES_RECORDS]);
82// SAFETY: Core 3 is the sole writer; readers bound indices via SERIES_COUNT Acquire.
83unsafe impl Sync for SyncSeriesRegistry {}
84
85static SERIES_RECORDS: SyncSeriesRegistry =
86    SyncSeriesRegistry([const { UnsafeCell::new(EMPTY_SERIES_RECORD) }; MAX_SERIES_RECORDS]);
87static SERIES_COUNT: AtomicUsize = AtomicUsize::new(0);
88
89#[derive(Debug, PartialEq, Eq)]
90pub enum DicomIngestError {
91    WorkerUnavailable,
92    Io(String),
93    Parse(String),
94    RegistryFull,
95    JobNotFound,
96    BlobRead(String),
97}
98
99impl std::fmt::Display for DicomIngestError {
100    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
101        match self {
102            Self::WorkerUnavailable => write!(f, "Core 3 DICOM worker not initialized"),
103            Self::Io(msg) => write!(f, "DICOM ingest IO: {msg}"),
104            Self::Parse(msg) => write!(f, "DICOM ingest parse: {msg}"),
105            Self::RegistryFull => write!(f, "DICOM series registry full"),
106            Self::JobNotFound => write!(f, "DICOM ingest job not found"),
107            Self::BlobRead(msg) => write!(f, "DICOM blob read: {msg}"),
108        }
109    }
110}
111
112impl std::error::Error for DicomIngestError {}
113
114/// Append-only pixel blob store (Core 3 writer).
115pub struct DicomBlobStore {
116    file: File,
117    write_offset: u64,
118}
119
120impl DicomBlobStore {
121    pub fn open(path: &Path) -> Result<Self, std::io::Error> {
122        let file = OpenOptions::new()
123            .read(true)
124            .write(true)
125            .create(true)
126            .open(path)?;
127        let len = file.metadata()?.len();
128        Ok(Self {
129            file,
130            write_offset: len,
131        })
132    }
133
134    pub fn append_pixels(&mut self, pixels: &[u8]) -> Result<u64, std::io::Error> {
135        let offset = self.write_offset;
136        self.file.seek(SeekFrom::Start(offset))?;
137        self.file.write_all(pixels)?;
138        self.file.sync_data()?;
139        self.write_offset = offset.saturating_add(pixels.len() as u64);
140        Ok(offset)
141    }
142
143    pub fn path_mmap(&self) -> Result<Mmap, std::io::Error> {
144        unsafe { memmap2::MmapOptions::new().map(&self.file) }
145    }
146}
147
148/// Zero-copy slice reader over the blob file (Core 1 / FRB boundary).
149pub struct DicomBlobReader {
150    mmap: Mmap,
151}
152
153impl DicomBlobReader {
154    pub fn open(path: &Path) -> Result<Self, DicomIngestError> {
155        let file = File::open(path).map_err(|e| DicomIngestError::Io(e.to_string()))?;
156        let mmap = unsafe { memmap2::MmapOptions::new().map(&file) }
157            .map_err(|e| DicomIngestError::Io(e.to_string()))?;
158        Ok(Self { mmap })
159    }
160
161    #[inline]
162    pub fn slice(&self, offset: usize, length: usize) -> Result<&[u8], DicomIngestError> {
163        let end = offset
164            .checked_add(length)
165            .ok_or_else(|| DicomIngestError::BlobRead("overflow".into()))?;
166        if end > self.mmap.len() {
167            return Err(DicomIngestError::BlobRead("range past blob end".into()));
168        }
169        Ok(&self.mmap[offset..end])
170    }
171}
172
173fn imaging_dir(storage_root: &Path) -> PathBuf {
174    storage_root.join("Imaging")
175}
176
177fn blob_path(storage_root: &Path) -> PathBuf {
178    imaging_dir(storage_root).join(BLOB_FILE_NAME)
179}
180
181fn wal_path(storage_root: &Path) -> PathBuf {
182    storage_root.join("qualia_global.wal")
183}
184
185fn hash_series_uid(uid: &str) -> u64 {
186    if uid.is_empty() {
187        return 0;
188    }
189    q_hash(uid)
190}
191
192fn infer_organ_hash(meta: &DicomMetadata) -> u64 {
193    if let Some(organ) =
194        crate::dicom::infer_organ_from_metadata(meta, &crate::dicom::default_organ_matchers())
195    {
196        q_hash(&organ)
197    } else {
198        0
199    }
200}
201
202fn compile_semantic_quins(
203    meta: &DicomMetadata,
204    patient_did_hash: u64,
205    series_hash: u64,
206    blob_offset: u64,
207    blob_length: u32,
208    out: &mut [NQuin; 6],
209) -> usize {
210    let study_ctx = q_hash("q42:imagingStudy");
211    let modality_hash = q_hash(&meta.modality);
212    let pred_has_series = q_hash("q42:hasDicomSeries");
213    let pred_modality = q_hash("q42:hasModality");
214    let pred_pixel_ptr = q_hash("q42:pixelBlobPointer");
215    let pred_body_part = q_hash("q42:bodyPartExamined");
216
217    let mut count = 0usize;
218
219    let mut q0 = NQuin::default();
220    q0.subject = patient_did_hash;
221    q0.predicate = pred_has_series;
222    q0.object = series_hash;
223    q0.context = study_ctx;
224    q0.metadata = pack_volume_metadata(meta.rows, meta.columns, blob_length);
225    q0.parity = q0.subject ^ q0.predicate ^ q0.object ^ q0.context;
226    out[count] = q0;
227    count += 1;
228
229    let mut q1 = NQuin::default();
230    q1.subject = series_hash;
231    q1.predicate = pred_modality;
232    q1.object = modality_hash;
233    q1.context = study_ctx;
234    q1.parity = q1.subject ^ q1.predicate ^ q1.object ^ q1.context;
235    out[count] = q1;
236    count += 1;
237
238    if !meta.body_part_examined.is_empty() {
239        let mut q2 = NQuin::default();
240        q2.subject = series_hash;
241        q2.predicate = pred_body_part;
242        q2.object = q_hash(&meta.body_part_examined);
243        q2.context = study_ctx;
244        q2.parity = q2.subject ^ q2.predicate ^ q2.object ^ q2.context;
245        out[count] = q2;
246        count += 1;
247    }
248
249    let mut q3 = NQuin::default();
250    q3.subject = series_hash;
251    q3.predicate = pred_pixel_ptr;
252    q3.object = encode_blob_pointer(blob_offset);
253    q3.context = study_ctx;
254    q3.metadata = pack_volume_metadata(meta.rows, meta.columns, blob_length);
255    q3.parity = q3.subject ^ q3.predicate ^ q3.object ^ q3.context;
256    out[count] = q3;
257    count += 1;
258
259    let _ = decode_blob_pointer(q3.object);
260    count
261}
262
263fn publish_series_record(record: DicomSeriesRecord) -> Result<usize, DicomIngestError> {
264    let idx = SERIES_COUNT.load(Ordering::Acquire);
265    if idx >= MAX_SERIES_RECORDS {
266        return Err(DicomIngestError::RegistryFull);
267    }
268    // SAFETY: single Core-3 writer; readers use Acquire on SERIES_COUNT before indexing.
269    unsafe {
270        *SERIES_RECORDS.0[idx].get() = record;
271    }
272    SERIES_COUNT.store(idx + 1, Ordering::Release);
273    Ok(idx)
274}
275
276fn run_split_ingest(job: IngestJob) {
277    let slot_idx = ((job.job_id as usize) - 1) % MAX_SERIES_RECORDS;
278    let slot = &JOB_SLOTS[slot_idx];
279
280    let result = (|| -> Result<DicomSeriesRecord, DicomIngestError> {
281        std::fs::create_dir_all(imaging_dir(&job.storage_root))
282            .map_err(|e| DicomIngestError::Io(e.to_string()))?;
283
284        let mut file_bytes = Vec::new();
285        File::open(&job.source_path)
286            .and_then(|mut f| f.read_to_end(&mut file_bytes))
287            .map_err(|e| DicomIngestError::Io(e.to_string()))?;
288
289        let DicomSplitPayload { meta, pixels } =
290            split_dicom_payload(&file_bytes).map_err(|e| DicomIngestError::Parse(e.to_string()))?;
291
292        let pixel_bytes = &file_bytes[pixels.offset..pixels.offset + pixels.length];
293        let mut store = DicomBlobStore::open(&blob_path(&job.storage_root))
294            .map_err(|e| DicomIngestError::Io(e.to_string()))?;
295        let blob_offset = store
296            .append_pixels(pixel_bytes)
297            .map_err(|e| DicomIngestError::Io(e.to_string()))?;
298
299        let series_hash = hash_series_uid(&meta.series_instance_uid);
300        let patient = if job.patient_did_hash != 0 {
301            job.patient_did_hash
302        } else if !meta.patient_id.is_empty() {
303            q_hash(&meta.patient_id)
304        } else {
305            q_hash("q42:anonymousPatient")
306        };
307
308        let mut quins = [NQuin::default(); 6];
309        let quin_count = compile_semantic_quins(
310            &meta,
311            patient,
312            series_hash,
313            blob_offset,
314            pixels.length as u32,
315            &mut quins,
316        );
317
318        let wal_file = wal_path(&job.storage_root);
319        if let Ok(mut wal) = WriteAheadLog::open(&wal_file) {
320            for quin in &quins[..quin_count] {
321                let _ = wal.append_mutation(quin);
322            }
323        }
324
325        let pixel_ptr_parity = quins[quin_count.saturating_sub(1)].parity;
326        Ok(DicomSeriesRecord {
327            patient_did_hash: patient,
328            series_hash,
329            blob_offset,
330            blob_length: pixels.length as u32,
331            rows: meta.rows,
332            cols: meta.columns,
333            organ_hash: infer_organ_hash(&meta),
334            pixel_pointer_quin_parity: pixel_ptr_parity,
335        })
336    })();
337
338    match result {
339        Ok(record) => {
340            slot.series_hash
341                .store(record.series_hash, Ordering::Release);
342            slot.blob_offset
343                .store(record.blob_offset, Ordering::Release);
344            slot.blob_length
345                .store(record.blob_length, Ordering::Release);
346            let _ = publish_series_record(record);
347            slot.status.store(JOB_COMPLETE, Ordering::Release);
348        }
349        Err(_) => {
350            slot.status.store(JOB_FAILED, Ordering::Release);
351        }
352    }
353}
354
355/// Pin the Core 3 DICOM Swarm worker (idempotent).
356pub fn init_core3_dicom_worker(storage_root: PathBuf) {
357    if IO_TX.get().is_some() {
358        return;
359    }
360
361    let (tx, rx): (Sender<IngestJob>, Receiver<IngestJob>) = bounded(32);
362    let _ = IO_TX.set(tx);
363
364    thread::Builder::new()
365        .name("qualia-core3-dicom".into())
366        .spawn(move || {
367            while let Ok(job) = rx.recv() {
368                run_split_ingest(job);
369            }
370        })
371        .expect("spawn Core 3 DICOM worker");
372
373    let _ = storage_root;
374}
375
376fn job_sender() -> Result<&'static Sender<IngestJob>, DicomIngestError> {
377    IO_TX.get().ok_or(DicomIngestError::WorkerUnavailable)
378}
379
380/// Submit a `.dcm` path to Core 3; returns a lock-free job id immediately.
381pub fn submit_dicom_ingest(
382    storage_root: &Path,
383    source_path: &Path,
384    patient_did_hash: u64,
385) -> Result<u64, DicomIngestError> {
386    init_core3_dicom_worker(storage_root.to_path_buf());
387    let job_id = NEXT_JOB_ID.fetch_add(1, Ordering::Relaxed);
388    let slot_idx = ((job_id as usize) - 1) % MAX_SERIES_RECORDS;
389    JOB_SLOTS[slot_idx]
390        .status
391        .store(JOB_PENDING, Ordering::Release);
392
393    job_sender()?
394        .send(IngestJob {
395            job_id,
396            source_path: source_path.to_path_buf(),
397            patient_did_hash,
398            storage_root: storage_root.to_path_buf(),
399        })
400        .map_err(|e| DicomIngestError::Io(e.to_string()))?;
401
402    Ok(job_id)
403}
404
405/// Poll job completion without blocking Core 1.
406pub fn dicom_ingest_status(job_id: u64) -> u8 {
407    let slot_idx = ((job_id as usize) - 1) % MAX_SERIES_RECORDS;
408    JOB_SLOTS[slot_idx].status.load(Ordering::Acquire)
409}
410
411/// Synchronous split-ingest for tests and CLI (runs on caller thread).
412pub fn split_ingest_sync(
413    storage_root: &Path,
414    source_path: &Path,
415    patient_did_hash: u64,
416) -> Result<DicomSeriesRecord, DicomIngestError> {
417    let mut file_bytes = Vec::new();
418    File::open(source_path)
419        .and_then(|mut f| f.read_to_end(&mut file_bytes))
420        .map_err(|e| DicomIngestError::Io(e.to_string()))?;
421
422    let DicomSplitPayload { meta, pixels } =
423        split_dicom_payload(&file_bytes).map_err(|e| DicomIngestError::Parse(e.to_string()))?;
424
425    std::fs::create_dir_all(imaging_dir(storage_root))
426        .map_err(|e| DicomIngestError::Io(e.to_string()))?;
427
428    let pixel_bytes = &file_bytes[pixels.offset..pixels.offset + pixels.length];
429    let mut store = DicomBlobStore::open(&blob_path(storage_root))
430        .map_err(|e| DicomIngestError::Io(e.to_string()))?;
431    let blob_offset = store
432        .append_pixels(pixel_bytes)
433        .map_err(|e| DicomIngestError::Io(e.to_string()))?;
434
435    let series_hash = hash_series_uid(&meta.series_instance_uid);
436    let patient = if patient_did_hash != 0 {
437        patient_did_hash
438    } else if !meta.patient_id.is_empty() {
439        q_hash(&meta.patient_id)
440    } else {
441        q_hash("q42:anonymousPatient")
442    };
443
444    let mut quins = [NQuin::default(); 6];
445    let quin_count = compile_semantic_quins(
446        &meta,
447        patient,
448        series_hash,
449        blob_offset,
450        pixels.length as u32,
451        &mut quins,
452    );
453
454    let wal_file = wal_path(storage_root);
455    if let Ok(mut wal) = WriteAheadLog::open(&wal_file) {
456        for quin in &quins[..quin_count] {
457            let _ = wal.append_mutation(quin);
458        }
459    }
460
461    let record = DicomSeriesRecord {
462        patient_did_hash: patient,
463        series_hash,
464        blob_offset,
465        blob_length: pixels.length as u32,
466        rows: meta.rows,
467        cols: meta.columns,
468        organ_hash: infer_organ_hash(&meta),
469        pixel_pointer_quin_parity: quins[quin_count.saturating_sub(1)].parity,
470    };
471    publish_series_record(record)?;
472    Ok(record)
473}
474
475#[cfg(test)]
476pub fn reset_ingest_registry_for_tests() {
477    SERIES_COUNT.store(0, Ordering::Release);
478}
479
480#[allow(dead_code)]
481fn split_ingest_sync_via_worker(
482    storage_root: &Path,
483    source_path: &Path,
484    patient_did_hash: u64,
485) -> Result<DicomSeriesRecord, DicomIngestError> {
486    init_core3_dicom_worker(storage_root.to_path_buf());
487    let job_id = NEXT_JOB_ID.fetch_add(1, Ordering::Relaxed);
488    run_split_ingest(IngestJob {
489        job_id,
490        source_path: source_path.to_path_buf(),
491        patient_did_hash,
492        storage_root: storage_root.to_path_buf(),
493    });
494    if dicom_ingest_status(job_id) != JOB_COMPLETE {
495        return Err(DicomIngestError::Parse("split ingest worker failed".into()));
496    }
497    let idx = SERIES_COUNT.load(Ordering::Acquire);
498    if idx == 0 {
499        return Err(DicomIngestError::JobNotFound);
500    }
501    Ok(unsafe { *SERIES_RECORDS.0[idx - 1].get() })
502}
503
504pub fn series_records_snapshot(out: &mut [DicomSeriesRecord]) -> usize {
505    let count = SERIES_COUNT.load(Ordering::Acquire).min(MAX_SERIES_RECORDS);
506    for (i, slot) in out.iter_mut().enumerate().take(count) {
507        // SAFETY: reader synchronizes via Acquire load on SERIES_COUNT.
508        *slot = unsafe { *SERIES_RECORDS.0[i].get() };
509    }
510    count
511}
512
513pub fn find_series_record(patient_did_hash: u64, series_hash: u64) -> Option<DicomSeriesRecord> {
514    let count = SERIES_COUNT.load(Ordering::Acquire);
515    for i in 0..count.min(MAX_SERIES_RECORDS) {
516        let record = unsafe { *SERIES_RECORDS.0[i].get() };
517        if record.patient_did_hash == patient_did_hash && record.series_hash == series_hash {
518            return Some(record);
519        }
520    }
521    None
522}
523
524/// Read a pixel payload slice from the blob store (single copy at FRB boundary).
525pub fn read_volume_bytes(
526    storage_root: &Path,
527    record: &DicomSeriesRecord,
528) -> Result<Vec<u8>, DicomIngestError> {
529    let reader = DicomBlobReader::open(&blob_path(storage_root))?;
530    Ok(reader
531        .slice(record.blob_offset as usize, record.blob_length as usize)?
532        .to_vec())
533}
534
535#[cfg(test)]
536mod tests {
537    use super::*;
538    use crate::dicom::{encode_blob_pointer, INLINE_TAG_BLOB_POINTER};
539    use tempfile::TempDir;
540
541    #[test]
542    fn blob_pointer_tag_encoding() {
543        let ptr = encode_blob_pointer(4096);
544        assert_eq!(ptr & INLINE_TAG_BLOB_POINTER, INLINE_TAG_BLOB_POINTER);
545        assert_eq!(crate::dicom::decode_blob_pointer(ptr), Some(4096));
546    }
547
548    /// End-to-end split ingest against gitignored private DICOM (skips if folder absent).
549    #[test]
550    fn local_private_dicom_split_ingest() {
551        let Some(root) = crate::dicom::resolve_local_dicom_dir() else {
552            eprintln!("skip: private DICOM fixtures not present");
553            return;
554        };
555        let paths = crate::dicom::collect_dicom_image_paths_under(&root, 2);
556        assert!(
557            !paths.is_empty(),
558            "no image slices under {}",
559            root.display()
560        );
561
562        reset_ingest_registry_for_tests();
563        let tmp = TempDir::new().unwrap();
564        let storage = tmp.path().to_path_buf();
565
566        for path in &paths {
567            let record =
568                split_ingest_sync(&storage, path, q_hash("did:patient:local-dicom-test")).unwrap();
569            assert!(record.blob_length > 0);
570            assert!(record.rows > 0);
571            assert!(record.cols > 0);
572            let blob = read_volume_bytes(&storage, &record).unwrap();
573            assert_eq!(blob.len(), record.blob_length as usize);
574        }
575    }
576
577    #[test]
578    fn split_ingest_writes_blob_and_registry() {
579        reset_ingest_registry_for_tests();
580        let tmp = TempDir::new().unwrap();
581        let storage = tmp.path().to_path_buf();
582        let bytes = crate::dicom::test_fixture_split_bytes();
583        let dcm_path = storage.join("slice.dcm");
584        std::fs::write(&dcm_path, &bytes).unwrap();
585
586        let split = crate::dicom::split_dicom_payload(&bytes).expect("fixture must parse");
587        assert_eq!(split.pixels.length, 4);
588
589        let record = split_ingest_sync(&storage, &dcm_path, q_hash("did:patient:test")).unwrap();
590        assert!(record.blob_length > 0);
591        assert_eq!(record.rows, 2);
592        assert_eq!(record.cols, 2);
593
594        let blob = read_volume_bytes(&storage, &record).unwrap();
595        assert_eq!(blob.len(), 4);
596        assert_eq!(blob[0], 10);
597    }
598}
qualia_core_db/medical/dicom_ingest.rs

qualia_core_db/medical/
dicom_ingest.rs
