Streaming Redaction

Overview

VeilData supports chunk-based streaming redaction, enabling efficient processing of large files and real-time streams while correctly detecting entities that span across chunk boundaries.

Features

• Streaming Buffer: Process text in chunks with minimal memory footprint

• Cross-Chunk Detection: Detect entities split across chunk boundaries using sliding window

• REST API: FastAPI endpoint for streaming HTTP requests/responses

• CLI Support: --stream flag for processing large files

• Reversible: Full TokenStore integration for reveal operations

Quick Start

Python API

from veildata.detectors import RegexDetector
from veildata.pipeline import DetectionPipeline
from veildata.streaming_buffer import StreamingRedactionBuffer

# Setup
patterns = {"EMAIL": r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}"}
detector = RegexDetector(patterns)
pipeline = DetectionPipeline(detector)
buffer = StreamingRedactionBuffer(pipeline, overlap_size=512)

# Process chunks
for chunk in ["My email is john.do", "[email protected] here"]:
    output = buffer.add_chunk(chunk)
    print(output, end="")

print(buffer.finalize())
# Output: My email is [REDACTED_1] here

CLI Streaming

# Stream large file
veildata redact large_file.txt --stream --output redacted.txt

# Custom chunk and overlap sizes
veildata redact input.txt --stream \
  --chunk-size 8192 \
  --overlap 1024 \
  --output output.txt \
  --store store.json

# With timing
veildata redact input.txt --stream --time --verbose

REST API

Start the API server:

pip install veildata[api]
uvicorn veildata.api.app:app --reload

Stream redaction:

curl -X POST "http://localhost:8000/v1/redact/stream?overlap_size=256" \
  -H "Content-Type: text/plain" \
  --data-binary @input.txt > output.txt

Get token store:

curl -X POST "http://localhost:8000/v1/redact/stream?return_store=true" \
  -H "Content-Type: text/plain" \
  --data-binary @input.txt \
  -v 2>&1 | grep x-token-store

How It Works

Sliding Window Algorithm

The streaming buffer uses a sliding window approach:

1. Buffer Accumulation: Chunks are appended to an internal buffer

2. Safe Zone Calculation: Text before the overlap region is “safe” to output

3. Cross-Boundary Detection: Entities in the overlap may span to next chunk

4. Progressive Output: Safe text is redacted and yielded immediately

5. Overlap Retention: Last N characters are kept for the next iteration

Chunk 1: "My email is john.do"
         |<---safe--->|<-overlap->|
         "My email is"   "john.do"  <- kept

Chunk 2: "[email protected] and more"
Buffer:  "[email protected] and more"
         |<-------safe-------->|<-overlap->|
Output: "[REDACTED_1] and"      "more"  <- kept

Finalize: "more"
Output: "more"

Entity Detection

Entities are detected using the full buffer (safe + overlap), but only entities entirely within the safe zone are redacted in the current iteration. This ensures:

• ✅ Entities split across chunks are detected when reassembled

• ✅ No partial entity text is output unredacted

• ✅ Minimal memory usage (only overlap region retained)

Configuration

Buffer Parameters

overlap_size (default: 512)

Characters to retain between chunks

• Should be ≥ longest expected entity

• Larger = better detection, more memory

• Smaller = less memory, may miss long entities

chunk_size (CLI only, default: 4096)

Bytes to read per chunk

• Affects I/O performance

• Does not affect detection accuracy

API Parameters

POST /v1/redact/stream

Query parameters:

• method: Redaction method (regex, ner_spacy, ner_bert)

• detect_mode: Detection mode (rules, ml, hybrid)

• overlap_size: Overlap for cross-chunk detection

• chunk_size: Request body chunk size

• return_store: Include token mappings in response header

Response headers:

• X-Redaction-Method: Method used

• X-Overlap-Size: Overlap size used

• X-Token-Store: JSON token mappings (if requested)

Performance

Benchmarks

Streaming mode provides constant memory usage regardless of file size:

File Size	Memory (Stream)	Memory (Batch)	Speedup
1 MB	~2 MB	~3 MB	1.0x
10 MB	~2 MB	~15 MB	0.95x
100 MB	~2 MB	~150 MB	0.9x
1 GB	~2 MB	OOM	∞

Note

Benchmarks with overlap_size=512, chunk_size=4096

When to Use Streaming

Use streaming when:

• ✅ Processing files > 100 MB

• ✅ Real-time stream processing

• ✅ Memory-constrained environments

• ✅ Need progressive output

Use batch mode when:

• ✅ Small files (< 10 MB)

• ✅ Maximum performance needed

• ✅ Need explain mode (not supported in streaming)

Limitations

Warning

• Explain mode: Not available in streaming (entities not tracked by position)

• Overlap size: Must be ≥ longest entity for reliable detection

• Performance: ~10% slower than batch for small files due to buffer overhead

Examples

See Examples for complete, working examples including:

• Basic buffer usage

• File streaming with progress tracking

• API integration

• CLI usage patterns

• Memory-efficient pipelines

Testing

Run streaming tests:

pytest tests/test_streaming_buffer.py -v    # Buffer tests (27 tests)
pytest tests/test_api.py -v                  # API tests (requires fastapi)

Architecture

Core Components

StreamingRedactionBuffer

Main buffer with sliding window logic

ChunkMetadata

Tracking for each processed chunk

stream_redact()

Convenience wrapper for generators

app.py

FastAPI application with streaming endpoint

Integration Points

• Compatible with all existing Detector implementations

• Uses standard DetectionPipeline for processing

• Full TokenStore support for reversible redaction

• Drop-in replacement for batch processing in CLI

API Reference

class veildata.streaming_buffer.StreamingRedactionBuffer(pipeline: DetectionPipeline, overlap_size: int = 512, store: TokenStore | None = None, redaction_format: str = '[REDACTED_{counter}]')

Bases: object

Buffer for streaming redaction with cross-chunk entity detection.

This class maintains a sliding window to ensure entities that span chunk boundaries are correctly detected and redacted. It progressively yields sanitized output while maintaining minimal state.

Example

>>> from veildata.detectors import RegexDetector
>>> from veildata.pipeline import DetectionPipeline
>>> detector = RegexDetector({"EMAIL": r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}"})
>>> pipeline = DetectionPipeline(detector)
>>> buffer = StreamingRedactionBuffer(pipeline, overlap_size=20)
>>> for chunk in ["My email is john.do", "[email protected] and more"]:
...     output = buffer.add_chunk(chunk)
...     print(output, end="")
>>> print(buffer.finalize())

Parameters:

• pipeline – DetectionPipeline to use for entity detection

• overlap_size – Number of characters to retain between chunks for boundary detection

• store – Optional TokenStore to record redacted mappings

• redaction_format – Format string for redaction tokens (default: “[REDACTED_{counter}]”)

add_chunk(chunk: str) → str

Add a chunk of text and return the redacted output for the safe zone.

The safe zone is the portion of the buffer that is guaranteed not to contain any entities that might continue in the next chunk.

Parameters:

chunk – New text chunk to process

Returns:

Redacted text for the safe portion of the buffer

finalize() → str

Process any remaining text in the buffer.

This should be called after all chunks have been added to ensure the overlap region is processed.

Returns:

Redacted text for the remaining buffer contents

get_metadata() → List[ChunkMetadata]

Get metadata about all processed chunks.

Returns:

List of ChunkMetadata objects

get_stats() → dict

Get statistics about the streaming process.

Returns:

Dictionary with processing statistics

reset() → None

Reset the buffer to initial state.

class veildata.streaming_buffer.ChunkMetadata(chunk_index: int, input_size: int, output_size: int, buffer_size: int, entities_detected: int)

Bases: object

Metadata about a processed chunk for debugging and tracking.

buffer_size: int

chunk_index: int

entities_detected: int

input_size: int

output_size: int

veildata.streaming_buffer.stream_redact(chunks: Generator[str, None, None], pipeline: DetectionPipeline, overlap_size: int = 512, store: TokenStore | None = None) → Generator[str, None, None]

Convenience function for streaming redaction.

This is a simple wrapper around StreamingRedactionBuffer for common use cases.

Parameters:

• chunks – Generator yielding text chunks

• pipeline – DetectionPipeline to use for entity detection

• overlap_size – Number of characters to retain between chunks

• store – Optional TokenStore to record redacted mappings

Yields:

Redacted text chunks

Example

>>> def read_file_chunks(filepath, chunk_size=4096):
...     with open(filepath, 'r') as f:
...         while True:
...             chunk = f.read(chunk_size)
...             if not chunk:
...                 break
...             yield chunk
>>>
>>> from veildata.engine import build_redactor
>>> redactor, store = build_redactor(method="regex")
>>> for redacted_chunk in stream_redact(read_file_chunks("input.txt"), redactor):
...     print(redacted_chunk, end="")

Future Enhancements

• Async streaming buffer for concurrent processing

• Adaptive overlap sizing based on detected entity lengths

• Streaming reveal endpoint

• WebSocket support for real-time streams

• Progress callbacks for large files

Examples

See examples/streaming_example.py for complete working examples.