CPOW Process Binary Analysis

Synapse CPOW-Process: Data Pipeline & Implementation Deep Dive

Location: /mnt/code/dev/Rust/energy-quotient/pq-synapse/server/crates/synapse-cpow/process

1. OVERVIEW & ENTRY POINTS

Binary Configuration

• Binary Name: cpow-process
• Source: src/bin/default.rs
• Features:
  • force_no_neon: Disables ARM NEON optimization (forces fallback implementation)
  • use_neon: Automatically enabled on ARM platforms (aarch64/arm)

Main Command-Line Architecture

cpow-process [OPTIONS] <COMMAND>

Global Options:
  --dev                    Enable development mode (custom shared memory naming)

Commands:
  default --verbose        Run the default wave processor
  parquet --output-dir     Parquet processor (future implementation)

Current Implementation Flow

The binary currently supports two execution paths:

1. Production: default command (–dev flag not set)
2. Development: default command with –dev flag

2. DATA PIPELINE ARCHITECTURE

High-Level Data Flow

Shared Memory (Raw Sensor Data)
           ↓
    SharedMemoryReader Thread
           ↓
  WaveBlockData → flume channel
           ↓
    Processor Thread
           ↓
   Data Conversion & Scaling
           ↓
Arrow RecordBatch Creation
           ↓
   Parquet File Writer
           ↓
    Parquet Files on Disk

Processing Stages

Stage 1: Shared Memory Reading (eq-memops::reader)

• Thread: Reader spawned with priority -10 (high priority on Linux)
• Input: SensorSharedMemory (circular buffer from CPOW acquire)
• Output: WaveBlockData tuples sent via flume channel
• Data Type: WaveBlockData = (Vec<SensorDataRow>, u64)
  • Vec<SensorDataRow>: 256 sensor samples (3 phases)
  • u64: Timestamp (nanoseconds since UNIX epoch)

Key Implementation (eq-memops/src/ipc/reader.rs:SharedMemoryReader::run):

#![allow(unused)]
fn main() {
// Reads N_ROWS_PER_BLOCK (256) rows at a time
// Block contains 7 channels × 256 rows of 24-bit data
// Timestamp marks start of block
// Detects overrun conditions and buffer state changes
}

Stage 2: Data Buffering & Overflow Handling

• Component: WaveSink (custom channel wrapper)
• Purpose: Manages bounded channel + overflow buffer + disk fallback
• Buffer States:
  • Normal: Data flowing normally
  • Buffering: Channel full, using in-memory overflow buffer
  • WritingToDisk: Overflow buffer full, writing to disk
• Configuration:
  • OVERFLOW_BUFFER_SIZE: 10 seconds of blocks (125 blocks)
  • MAX_FILE_SIZE: 32 MB per overflow file
  • MAX_TOTAL_SIZE: 512 MB total overflow storage

Stage 3: Processing & Format Conversion

• Thread: Processor spawned with priority -10
• Input: WaveBlockData from channel (flume receiver)
• Processing Engine: DefaultWaveProcessorType

Key Transformations:

1. 24-bit to 32-bit Conversion:

   #![allow(unused)]
   fn main() {
   // Each SensorDataRow contains 7 channels of 24-bit data
   // Converted to i32 (sign-extended)
   fn i24_to_i32(bytes: [u8; 3]) -> i32 {
       if (bytes[2] & 0x80) != 0 {  // Check sign bit
           (0xFF000000u32 | ...) as i32  // Negative (extend)
       } else {
           ((bytes[2] as u32) << 16 | ...) as i32  // Positive
       }
   }
   }

2. Data Batching (DataBuffer from eq-memops::batch):

   • Accumulates raw 24-bit data into 32-bit format
   • Organizes by channel
   • Batch size: configurable (default: 320,000 rows = 10 seconds @ 32 kHz)

3. Scaling & Arrow Conversion:

   • Current/voltage scaling factors applied
   • Conversion to Arrow arrays for columnar storage
   • Schema: [IA, VA, IB, VB, IC, VC, IN] (7 Int32 fields)

Stage 4: Parquet File Writing

• Component: ArrowIpc + ParquetWriter
• Output Location: $CPOW_DATA_DIR (from config)
• File Format: Parquet with Zstandard compression
• Metadata Storage:
  • start_time: RFC3339 timestamp with nanosecond precision
  • vscale: Voltage scaling factor
  • iscale: Current scaling factor

File Rotation Logic:

• Triggers when:
  • Row count exceeds 1,920,000 rows (~60 seconds @ 32 kHz)
  • 60 seconds elapsed (configurable)
  • Timestamp discontinuity > 1 second detected

3. SHARED MEMORY INTERFACE

SensorSharedMemory Structure (eq-cpow/sensor)

Memory Layout (cache-line aligned):

┌─────────────────────────────┐
│  SensorSharedControl (64B)  │  
│  - read_pos (atomic)        │
│  - write_pos (atomic)       │
│  - terminate_flag           │
│  - reader_active            │
│  - generation               │
├─────────────────────────────┤
│  SensorSharedData Header    │
│  - num_rows                 │
│  - vscale (f32)             │
│  - iscale (f32)             │
├─────────────────────────────┤
│  SensorDataRow[num_rows]    │  (21 bytes each, 256 rows/block)
│  ia, va, ib, vb, ic, vc, in│
│  (3 bytes each channel)     │
├─────────────────────────────┤
│  Timestamps[num_rows/256]   │  (8 bytes each, 1 per block)
└─────────────────────────────┘

Data Types

SensorDataRow (21 bytes, packed):

#![allow(unused)]
fn main() {
#[repr(C, packed)]
pub struct SensorDataRow {
    pub ia: [u8; 3],   // Phase A current
    pub va: [u8; 3],   // Phase A voltage
    pub ib: [u8; 3],   // Phase B current
    pub vb: [u8; 3],   // Phase B voltage
    pub ic: [u8; 3],   // Phase C current
    pub vc: [u8; 3],   // Phase C voltage
    pub i_n: [u8; 3],  // Neutral current
}
}

Key Constants:

• N_ROWS_PER_BLOCK: 256 (= 1 frame × 64 rows/frame)
• BLOCK_SIZE: 5,376 bytes (256 rows × 21 bytes)
• BLOCK_DURATION: 8 milliseconds @ 32 kHz
• SAMPLE_HZ: 32,000 Hz
• N_CHANNELS: 7

Reading Process

#![allow(unused)]
fn main() {
// In SharedMemoryReader::run()
loop {
    let rows_available = write_pos - read_pos;
    
    if rows_available >= N_ROWS_PER_BLOCK {
        // 1. Prepare data at read position
        let prepared_data = shared_mem.prepare_data_at(read_pos, N_ROWS_PER_BLOCK)?;
        // Returns: (Vec<SensorDataRow>, timestamp)
        
        // 2. Send to sink (channel)
        sink.try_consume(prepared_data)?;
        
        // 3. Advance read position
        read_pos = (read_pos + N_ROWS_PER_BLOCK) % num_rows;
    } else {
        sleep(Duration::from_micros(10));
    }
}
}

Buffer State Detection (metrics tracking):

• Normal: Standard operation
• OverrunRiskWarning: Buffer > 50% full (wrapped case)
• Overrun: Reader hasn’t kept up, data lost
• WritingToDisk: Overflow buffer being persisted

4. ARROW SHARED MEMORY IMPLEMENTATION (Partially Complete)

Status: TODO - Incomplete

Location: eq-memops/src/ipc/arrow.rs (lines 364+)

Architecture Design

ArrowSharedMemory Structure (proposed):

#![allow(unused)]
fn main() {
pub struct ArrowSharedMemory {
    shmem: Arc<Shmem>,
    schema: SchemaRef,
    control: *mut ArrowSharedControl,
    data_offset: usize,
    total_size: usize,
    reader_mode: ReaderMode,  // Single or Multi(MultiReaderState)
}
}

Control Structure (for lock-free multi-reader):

#![allow(unused)]
fn main() {
#[repr(align(64))]
pub struct ArrowSharedControl {
    write_pos: AtomicUsize,
    generation: AtomicI64,
    batch_infos: [BatchInfo; 1024],  // Up to 1024 batches
}

struct BatchInfo {
    offset: AtomicUsize,
    size: AtomicUsize,
    timestamp: AtomicI64,
}
}

Reader Control (multi-reader support):

#![allow(unused)]
fn main() {
#[repr(align(64))]
struct ReaderControl {
    positions: [AtomicUsize; 16],    // Up to 16 readers
    active: [AtomicUsize; 16],       // Reader active flags
}
}

Design Goals

1. Lock-free operation: Uses compare-and-swap for coordination
2. Multi-reader support: 16 concurrent readers maximum
3. Arrow Record Batch streaming: Direct serialization to shared memory
4. Circular buffer: Automatic wrapping when full

Implemented Methods

• new(): Create/open shared memory region with schema
• write(): Write RecordBatch as Arrow IPC format
• read(): Read batch by reader ID
• register_reader(): Register new reader (returns ID)
• unregister_reader(): Deactivate reader

Known Issues (TODO markers)

#![allow(unused)]
fn main() {
todo!("Determine intended logic and fix inst of ArrowSharedMemory");  // Line 364
}

The initialization logic is incomplete - control pointer setup and boundary calculations need finalization.

Proposed Use Cases (Future)

1. Live data streaming to web frontends via WebSocket
2. Real-time analytics consumer processes
3. Event detection system reading raw data simultaneously with file writing

5. NEON OPTIMIZATION (ARM-Specific)

Overview

The system uses ARM NEON (Single Instruction Multiple Data) instructions for high-performance 24-bit to 32-bit conversion and scaling.

Implementation

Location: eq-memops/src/neon_ops/

Rust Wrapper (conversion.rs):

#![allow(unused)]
fn main() {
pub fn i32_to_f32_scaled(input: &[i32], scale: f32) -> Vec<f32> {
    #[cfg(any(target_arch = "arm", target_arch = "aarch64"))]
    {
        convert_i32_to_f32_neon(input, scale)
    }

    #[cfg(not(any(target_arch = "arm", target_arch = "aarch64")))]
    {
        // Fallback for x86/other platforms
        input.iter().map(|&x| x as f32 * scale).collect()
    }
}
}

C Implementation (conversion.c):

• Uses ARM NEON intrinsics for batch conversion
• In-place transformation of i32 buffer to f32
• Handles scaling factor application

Build Integration (build.rs):

#![allow(unused)]
fn main() {
if (target.contains("aarch64") || target.contains("arm")) && !force_no_neon {
    cc::Build::new()
        .file("../../eq-memops/src/neon_ops/conversion.c")
        .compile("neon_ops");
    println!("cargo:rustc-cfg=feature=\"use_neon\"");
}
}

Force Disable:

cargo build --features force_no_neon

6. COMMAND-LINE OPTIONS & DEFAULT FLAG

Currently Supported Commands

default Command

cargo run --bin synapse-cpow-process -- default [OPTIONS]

Options:
  -v, --verbose    Enable verbose output during processing

Behavior:

• Reads from production shared memory path (no user suffix)
• Threads spawned with priority -10
• Metrics enabled by default
• Normal error handling with automatic recovery

default with --dev Flag

cargo run --bin synapse-cpow-process -- --dev default --verbose

Options:
  --dev                    Add user suffix to shared memory name
  -v, --verbose            Verbose mode in processor

Behavior:

• Loads config from database (SynapseConfig::from_db_sync())
• Appends _{USER} to shared memory name
• Example: raw_cpow_shmem_user instead of raw_cpow_shmem
• Useful for development without interfering with production data

Future --default Flag (Not Yet Implemented)

The code structure suggests future support for a --default flag to select default processor behavior:

#![allow(unused)]
fn main() {
#[derive(Subcommand, Debug)]
enum ProcessorCommands {
    Default { .. },      // Current implementation
    Parquet { .. },      // Future: custom parquet output location
}
}

This would allow:

# Potential future syntax
cargo run --bin synapse-cpow-process -- --default

7. PROCESSING PIPELINE DETAILED WALKTHROUGH

Thread Initialization

Main Thread (src/bin/default.rs:main()):

#![allow(unused)]
fn main() {
// 1. Create channel for data flow
let (tx, rx) = flume::unbounded::<WaveBlockData>();

// 2. Create WaveSink with overflow handling
let sink = WaveSink::new(tx, OVERFLOW_BUFFER_SIZE)?;

// 3. Spawn reader thread
let reader_handle = spawn_thread::<_, _, _, WaveProcessError>(
    "shared_memory_reader", 
    move || {
        // Set priority to -10 (high)
        setpriority(PRIO_PROCESS, 0, -10);
        
        // Build and run reader
        let mut reader = SharedMemoryReaderBuilder::<
            SensorSharedMemory, 
            WaveSink, 
            WaveBlockData
        >::default(sink)?
            .with_metrics(true)
            .build()?;
        
        reader.run(Arc::clone(&term_flag_ref))
    }
)?;

// 4. Spawn processor thread
let processor = ProcessorBuilder::new()
    .with_source(rx)
    .build()?;

let mut proc_runner = processor.into_runner()?;

let proc_handle = spawn_thread::<_, _, _, WaveProcessError>(
    "processor",
    move || {
        setpriority(PRIO_PROCESS, 0, -10);
        proc_runner.run_as::<DefaultWaveProcessorType>(
            Arc::clone(&terminate_flag)
        )
    }
)?;
}

Reader Thread Operation

Location: eq-memops/src/ipc/reader.rs:SharedMemoryReader::run()

#![allow(unused)]
fn main() {
// 1. Wait for writer to become active
while !terminate_flag.load(Ordering::SeqCst) {
    if self.shared_mem.is_writer_active() { break; }
    thread::sleep(Duration::from_secs(1));
}

// 2. Initialize read position (align to block boundary)
let mut read_pos = self.shared_mem.read_position();
read_pos = read_pos - (read_pos % N_ROWS_PER_BLOCK);
self.shared_mem.set_read_position(read_pos);

// 3. Main read loop
loop {
    let write_pos = self.shared_mem.write_position();
    let rows_available = (write_pos + num_rows - read_pos) % num_rows;
    
    if rows_available >= N_ROWS_PER_BLOCK {
        // Prepare data block
        let prepared_data = self.shared_mem.prepare_data_at(
            read_pos, 
            N_ROWS_PER_BLOCK
        )?;
        
        // Send to sink
        match self.sink.try_consume(prepared_data) {
            Ok(()) => {
                read_pos = (read_pos + N_ROWS_PER_BLOCK) % num_rows;
                self.shared_mem.set_read_position(read_pos);
            }
            Err(_) => {
                // Handle overflow - written to buffer/disk
            }
        }
    } else {
        thread::sleep(Duration::from_micros(10));
    }
    
    if terminate_flag.load(Ordering::Acquire) { break; }
}
}

Processor Thread Operation

Location: eq-processing/src/default.rs:DefaultWaveProcessorType::run()

#![allow(unused)]
fn main() {
// 1. Initialize Arrow IPC writer with schema
let schema = Arc::new(Schema::new(vec![
    Field::new("IA", DataType::Int32, false),
    Field::new("VA", DataType::Int32, false),
    // ... 7 channels total
]));

let mut arrow_ipc = ArrowIpc::new(schema, dest, scaler)?;
arrow_ipc.initialize_writer()?;

// 2. Main processing loop
loop {
    // Check if rotation needed (every 1 second)
    if last_rotate_check.elapsed() >= Duration::from_secs(1) {
        if arrow_ipc.check_rotation_needed()? {
            arrow_ipc.rotate()?;  // Close file, open new one
        }
        last_rotate_check = std::time::Instant::now();
    }
    
    // Receive block from channel
    match self.processor.source.recv() {
        Ok((rows, timestamp)) => {
            // Create buffer and convert to Arrow format
            let mut buffer = DataBuffer::new(rows.len());
            buffer.push_datarows(&rows)?;
            
            let mut view = buffer.raw_view(buffer.len())?;
            
            // Get first batch
            if let Some(batch) = view.next_result()? {
                // Check for timestamp discontinuity (>1s gap)
                if timestamp > (last_ts + 1_000_000_000) {
                    tracing::warn!("Timestamp discontinuity - rotating file");
                    arrow_ipc.rotate()?;
                }
                
                // Write to parquet
                arrow_ipc.write_batch(&batch, timestamp)?;
                last_ts = timestamp;
            }
        }
        Err(ChannelError::Empty) => {
            thread::sleep(Duration::from_micros(10));
        }
        Err(ChannelError::Disconnected) => break,
    }
    
    if terminate_flag.load(Ordering::Acquire) { break; }
}
}

Data Conversion Pipeline

24-bit to 32-bit in DataBuffer:

#![allow(unused)]
fn main() {
pub fn push_datarows(&mut self, rows: &[SensorDataRow]) -> Result<usize, BatchError> {
    for row in rows {
        let offset = self.write_pos * N_CHANNELS;
        
        // Unroll field access - direct 24→32 conversion
        self.data[offset]     = i24_to_i32(row.ia);  // Ch0: IA
        self.data[offset + 1] = i24_to_i32(row.va);  // Ch1: VA
        self.data[offset + 2] = i24_to_i32(row.ib);  // Ch2: IB
        self.data[offset + 3] = i24_to_i32(row.vb);  // Ch3: VB
        self.data[offset + 4] = i24_to_i32(row.ic);  // Ch4: IC
        self.data[offset + 5] = i24_to_i32(row.vc);  // Ch5: VC
        self.data[offset + 6] = i24_to_i32(row.i_n); // Ch6: IN
        
        self.write_pos = (self.write_pos + 1) % self.capacity;
        self.len += 1;
    }
    Ok(n_rows)
}
}

Arrow RecordBatch Creation:

#![allow(unused)]
fn main() {
impl ProcessorSink<WaveBlockData> for ArrowIpc<...> {
    fn write_batch(&mut self, batch: &BatchSlice<i32>, timestamp: u64) {
        // Apply scaling to each channel
        let channel_arrays: Vec<Arc<dyn Array>> = batch
            .channels
            .par_iter()
            .enumerate()
            .map(|(channel_idx, channel_data)| {
                let scale_factor = self.scaler.scale_factor(channel_idx);
                
                // Scale: current (i%2==0) or voltage (i%2==1)
                let scaled_values: Vec<f32> = channel_data
                    .iter()
                    .map(|&v| scale_factor * v as f32)
                    .collect();
                
                Arc::new(arrow::array::Float32Array::from(scaled_values))
                    as Arc<dyn Array>
            })
            .collect();
        
        // Create batch and write to parquet
        let record_batch = RecordBatch::try_new(
            self.schema.clone(), 
            channel_arrays
        )?;
        
        self.writer.write_batch(&record_batch)?;
    }
}
}

8. PARQUET OUTPUT CONFIGURATION

File Format & Compression

Writer Configuration (ParquetWriter::init):

#![allow(unused)]
fn main() {
let props = WriterProperties::builder()
    .set_encoding(Encoding::DELTA_BINARY_PACKED)
    .set_compression(Compression::from(config.cpow.parquet.compression))
    .set_max_row_group_size(ROWS_PER_BATCH)
    .set_data_page_row_count_limit(ROWS_PER_BATCH / 4)
    .set_data_page_size_limit(PARQUET_DATA_PAGE_SIZE_LIMIT)
    .set_dictionary_page_size_limit(PARQUET_DICTIONARY_PAGE_SIZE_LIMIT)
    .set_statistics_enabled(EnabledStatistics::Page)
    .set_key_value_metadata(Some(metadata))
    .build();
}

Constants (ParquetConstants):

• ROWS_PER_BATCH: 320,000 rows (10 seconds @ 32 kHz)
• TERMINATION_CHECK_ROWS: 3,200 rows (100 ms)
• PARQUET_DATA_PAGE_SIZE_LIMIT: 83,886,080 bytes (~80 MB)
• PARQUET_DICTIONARY_PAGE_SIZE_LIMIT: 16,384 bytes (~16 KB)

File Structure

Filename Format:

{CPOW_DATA_DIR}/{YYYYMMDD_HHMMSS}.parquet

Metadata Key-Value Pairs:

• start_time: RFC3339 timestamp (nanosecond precision)
• vscale: Voltage scale factor (float)
• iscale: Current scale factor (float)

Row Groups:

• Size: 320,000 rows per group
• Compression: Zstandard (configurable)
• Encoding: Delta Binary Packed for integers

9. ERROR HANDLING & RECOVERY

Error Types

WaveProcessError hierarchy:

├─ Processing: Processor error
├─ Memops: Memory operations error
├─ Ipc: Inter-process communication error
├─ SharedMemory: Shared memory initialization error
├─ SharedMemoryReader: Reader-specific error
├─ Sensor: Sensor error
├─ Parquet: Parquet file error
├─ LedControl: LED control error
├─ DiskManager: Disk management error
├─ Channel: Custom channel error
│  ├─ Disconnected: Channel severed
│  ├─ Full: Channel at capacity
│  ├─ Timeout: Read timeout
│  ├─ BufferOverflow: Overflow buffer full
│  └─ DiskWrite: Failed to write overflow data
├─ Thread: Thread spawning error
├─ ThreadExecution: Thread runtime error
├─ Initialization: Startup error
├─ Configuration: Config loading error
├─ Io: Standard I/O error
└─ Other: Generic error

Overflow Handling Sequence

Normal Channel Send
    ↓
Channel Full? → Yes → handle_overflow()
    ↓ No
    Success

Overflow Handling:
  1. Try to pop from overflow buffer → primary channel
  2. If buffer not full → add block to buffer
  3. If buffer full → try_send to disk writer

Metrics & Monitoring

Reader Metrics (ReaderMetrics):

• blocks_processed: Total blocks read
• last_read_time: UNIX timestamp of last read
• shared_mem_buffer_state: Current shared memory state
• sink_buffer_state: Current channel state
• overflow_count: Times channel was full
• read_errors: Total read failures
• processing_rate: Blocks per second

10. PERFORMANCE CONSIDERATIONS

Thread Priorities

• Both reader and processor threads set to priority -10 (Linux)
• Higher priority = lower latency, prevents preemption

Memory Alignment

• SensorSharedControl: 64-byte cache-line aligned
• SensorSharedData: 64-byte cache-line aligned
• Minimizes false sharing between reader and processor

Batching Strategy

• Reader: Reads 256 rows per transaction (8 ms)
• Processor: Batches to 320,000 rows (10 seconds) before writing
• Channel: Unbounded primary + 1,250-row overflow buffer

NEON Optimization Impact

• ARM platforms: 4x+ speedup on 24→32-bit conversion
• Fallback: Automatic scalar processing on other architectures
• Force disable: Use force_no_neon feature for benchmarking

File I/O Strategy

• Asynchronous rotation: Checks every 1 second
• Triggers: Row count threshold OR time-based (60 sec)
• Timestamp discontinuity: Immediate rotation on > 1 second gap
• Compression: Zstandard (configurable level)

11. CONFIGURATION INTEGRATION

SynapseConfig Integration

#![allow(unused)]
fn main() {
// From config database
let config = SynapseConfig::from_db_sync()?;

// Used for:
- config.cpow.processing.shared_memory.raw.name
- config.cpow.parquet.compression
- Event detection parameters
- Processing thresholds
}

Development Mode User Suffix

#![allow(unused)]
fn main() {
let raw_shmem_name = if parsed.dev {
    let user = std::env::var("USER").unwrap_or_else(|_| "user".to_string());
    format!("{}_{}", config.cpow.processing.shared_memory.raw.name, user)
} else {
    config.cpow.processing.shared_memory.raw.name.clone()
};
}

12. KEY FILES & MODULES REFERENCE

13. SUMMARY: DATA FLOW WITH NUMBERS

Sensor @ 32 kHz → 7 channels × 24-bit data
        ↓
Shared Memory (circular buffer, 256-row blocks)
        ↓
Reader Thread (block every 8 ms, async priority -10)
        ↓
256 rows/block → WaveBlockData tuple
        ↓
Flume Channel (unbounded + 10-sec overflow buffer)
        ↓
Processor Thread (async priority -10)
        ↓
24-bit → 32-bit conversion (NEON accelerated on ARM)
        ↓
320,000-row batch accumulation (10 seconds)
        ↓
Arrow RecordBatch with scaling applied
        ↓
Parquet File Writing (Zstandard compression)
        ↓
1,920,000-row file limit OR 60-second rotation
        ↓
Timestamped Parquet files on disk (~YYYYMMDD_HHMMSS.parquet)

14. FUTURE ENHANCEMENTS & TODOS

1. Arrow Shared Memory (Partially Complete)

   • Complete ArrowSharedMemory::new() initialization
   • Implement multi-reader batch consumption
   • Integration with web services

2. Parquet Processor Command

   • Custom output directory support
   • Alternative compression strategies
   • File naming conventions

3. Metrics Enhancement

   • Export to Prometheus
   • Real-time performance dashboard
   • Per-channel statistics

4. Overflow Handler Improvements

   • Memory-mapped files instead of standard I/O
   • Compression for overflow data
   • Automatic cleanup of old overflow files

5. Configuration Enhancements

   • Per-command configuration overrides
   • Runtime parameter adjustment
   • Hot-reload support