This document describes the internal architecture, algorithms, and design decisions of Unirust. It serves as the authoritative reference for understanding how the system works.
- System Overview
- Core Concepts
- Entity Resolution Algorithm
- Conflict Detection
- Distributed Architecture
- Cross-Shard Reconciliation
- Storage Layer
- Performance Optimizations
- Data Flow
Unirust is a temporal entity resolution engine that clusters records from multiple source systems into unified master entities. The system supports both single-node and distributed deployments.
┌─────────────────┐
│ Clients │
│ (gRPC/API) │
└────────┬────────┘
│
┌────────▼────────┐
│ Router │
│ (hash-based │
│ routing) │
└────────┬────────┘
│
┌────────────────────┼────────────────────┐
│ │ │
┌───────▼───────┐ ┌───────▼───────┐ ┌───────▼───────┐
│ Shard 0 │ │ Shard 1 │ │ Shard N │
│ ┌─────────┐ │ │ ┌─────────┐ │ │ ┌─────────┐ │
│ │ Linker │ │ │ │ Linker │ │ │ │ Linker │ │
│ │ DSU │ │ │ │ DSU │ │ │ │ DSU │ │
│ │ Index │ │ │ │ Index │ │ │ │ Index │ │
│ │ RocksDB │ │ │ │ RocksDB │ │ │ │ RocksDB │ │
│ └─────────┘ │ │ └─────────┘ │ │ └─────────┘ │
└───────────────┘ └───────────────┘ └───────────────┘
A Record represents a single observation from a source system:
struct Record {
id: RecordId, // Unique within shard
identity: RecordIdentity, // Entity type + perspective + UID
descriptors: Vec<Descriptor>, // Attribute-value pairs with time intervals
}
struct RecordIdentity {
entity_type: String, // e.g., "person", "company"
perspective: String, // Source system, e.g., "crm", "erp"
uid: String, // Source-unique identifier
}
struct Descriptor {
attr: AttrId, // Interned attribute name
value: ValueId, // Interned value
interval: Interval, // [start, end) validity period
}The Ontology defines matching rules:
-
Identity Keys: Attribute combinations that identify the same entity
// Records with matching (name, email) are considered the same entity IdentityKey::new(vec![name_attr, email_attr], "name_email")
-
Strong Identifiers: Attributes that cannot conflict within a cluster
// SSN must be unique - conflicting SSNs block merges StrongIdentifier::new(ssn_attr, "ssn_unique")
-
Constraints: Validation rules
Constraint::unique(email_attr, "unique_email") Constraint::unique_within_perspective(account_id, "source_unique")
All data has temporal validity. An Interval [start, end) defines when a descriptor is valid. Entity resolution respects these intervals:
- Records only merge if their identity key values match during overlapping time periods
- Conflicts are detected per-interval, not globally
- Golden records are computed for each unique time period
The core algorithm uses batch-parallel processing for high throughput. Within each partition, records are processed in three phases:
Phase 1: Batch Store Insertion
│ All records added to store, collecting (index, record_id) pairs
▼
Phase 2: Parallel Extraction (Rayon)
│ Extract identity keys, strong ID summaries across all records
▼
Phase 3: Sequential Linking
│ DSU merges with temporal guards (serialized for correctness)
▼
Result: Cluster assignments returned in original batch order
Parallel extraction dominates compute time; DSU mutations are inherently sequential due to data dependencies but benefit from root caching.
For each record, extract values for all identity keys defined in the ontology:
fn extract_key_values(record: &Record, ontology: &Ontology) -> Vec<KeyValue> {
ontology.identity_keys()
.iter()
.filter_map(|key| {
// Collect all attribute values for this identity key
let values: Vec<_> = key.attributes()
.iter()
.filter_map(|attr| record.value_for(*attr))
.collect();
// Only complete keys (all attributes present) form valid key values
if values.len() == key.attributes().len() {
Some(KeyValue::new(key.name(), values))
} else {
None
}
})
.collect()
}For each key value, query the identity index:
fn find_candidates(key_value: &KeyValue, index: &IdentityIndex) -> Vec<RecordId> {
let signature = hash_key_value(key_value);
index.get(&signature).unwrap_or_default()
}The identity index maps key value signatures to record IDs. This is the hot path, optimized with:
- Bloom filters for fast negative lookups (16MB filter, <1% false positive rate)
- Sharded caching to avoid lock contention
- SIMD-accelerated hashing
The Disjoint Set Union (DSU) data structure tracks cluster membership. Merging follows these rules:
fn try_merge(dsu: &mut TemporalDSU, a: RecordId, b: RecordId,
store: &Store, ontology: &Ontology) -> MergeResult {
let root_a = dsu.find(a);
let root_b = dsu.find(b);
if root_a == root_b {
return MergeResult::AlreadySame;
}
// Check temporal guards
let guard = compute_temporal_guard(root_a, root_b, store, ontology);
match guard {
TemporalGuard::Allowed { reason, interval } => {
dsu.union_with_guard(root_a, root_b, guard);
MergeResult::Merged { interval }
}
TemporalGuard::Blocked { reason } => {
MergeResult::Conflict { reason }
}
}
}Temporal Guards validate merges:
- Overlapping Intervals: Records must have overlapping validity periods
- Strong Identifier Agreement: Strong identifiers must match during overlap
- Constraint Satisfaction: Uniqueness constraints must be satisfied
Each record receives its final cluster ID:
fn finalize_assignment(record_id: RecordId, dsu: &TemporalDSU) -> ClusterId {
ClusterId(dsu.find(record_id).0)
}To prevent pathological cases (e.g., very common names), candidate discovery uses adaptive capping:
| Candidate Count | Cap Applied |
|---|---|
| < 2,000 | No cap |
| 2,000 - 10,000 | Cap at 1,000 |
| > 10,000 | Cap at 500 |
| > 50,000 | Early exit (hot key) |
Additionally, stochastic sampling maintains match quality: when candidates exceed the threshold, random sampling weighted by temporal overlap preserves expected accuracy.
Conflicts occur when records in the same cluster have incompatible values for the same attribute during overlapping time periods.
Unirust implements two conflict detection algorithms with automatic selection:
Best for clusters with diverse time boundaries.
Events: [(t1, START, r1), (t2, END, r1), (t3, START, r2), ...]
sorted by time
Active set: records currently "open"
For each event:
if START: add to active set, check conflicts with all active
if END: remove from active set
Best for clusters with high overlap (many records share same intervals).
1. Collect all unique time boundaries: {t1, t2, t3, ...}
2. Create atomic intervals: [t1,t2), [t2,t3), [t3,t4), ...
3. For each atomic interval:
- Find all records active during this interval
- Group by attribute
- If multiple values for same attribute → conflict
fn select_algorithm(unique_boundaries: usize, total_descriptors: usize) -> Algorithm {
let max_boundaries = total_descriptors * 2; // Each descriptor has start + end
let ratio = unique_boundaries as f64 / max_boundaries as f64;
if ratio < 0.5 {
// High overlap → atomic intervals is faster
Algorithm::AtomicIntervals
} else {
// Low overlap → sweep line is faster
Algorithm::SweepLine
}
}-
Direct Conflict: Same attribute has different values in overlapping intervals
Record A: email = "john@foo.com" [100, 200) Record B: email = "john@bar.com" [150, 250) → Conflict in [150, 200) -
Indirect Conflict: Strong identifier violation
Record A: ssn = "123-45-6789" [100, 200) Record B: ssn = "987-65-4321" [150, 250) → Blocked merge (strong identifier conflict)
The router provides the external API and routes requests to shards:
impl Router {
async fn ingest(&self, records: Vec<Record>) -> Vec<Assignment> {
// Group records by target shard
let mut shard_batches: HashMap<ShardId, Vec<Record>> = HashMap::new();
for record in records {
let shard_id = self.route(&record);
shard_batches.entry(shard_id).or_default().push(record);
}
// Fan out to shards in parallel
let futures: Vec<_> = shard_batches
.into_iter()
.map(|(shard_id, batch)| {
self.shards[shard_id].ingest(batch)
})
.collect();
// Collect results
join_all(futures).await.into_iter().flatten().collect()
}
}Records are routed by hashing their identity key values:
fn route(record: &Record, num_shards: usize) -> ShardId {
let key_values = extract_key_values(record);
let hash = hash_key_values(&key_values);
ShardId(hash % num_shards as u64)
}This ensures records that might need to merge are routed to the same shard.
Each cluster has a globally unique ID:
┌──────────────────────────────────────────────────────────────┐
│ GlobalClusterId (64 bits) │
├──────────────┬────────────────────┬──────────────────────────┤
│ shard_id │ version │ local_id │
│ (16 bits) │ (16 bits) │ (32 bits) │
└──────────────┴────────────────────┴──────────────────────────┘
- shard_id: Owning shard (0-65535)
- version: Merge version for conflict detection
- local_id: Cluster ID within the shard
When records that should merge are routed to different shards, we need cross-shard reconciliation:
Shard 0 Shard 1
┌────────────┐ ┌────────────┐
│ Record A │ │ Record B │
│ name=John │ Should merge │ name=John │
│ email=j@x │ ◄───────────────► │ email=j@x │
│ Cluster 0 │ │ Cluster 5 │
└────────────┘ └────────────┘
Each shard maintains a Cluster Boundary Index tracking identity keys that appear:
struct ClusterBoundaryIndex {
// Maps identity key signature → boundary entries
boundaries: HashMap<IdentityKeySignature, Vec<BoundaryEntry>>,
// Bloom filter for fast negative lookups
bloom: BloomFilter,
// Keys modified since last reconciliation
dirty_keys: HashSet<IdentityKeySignature>,
}
struct BoundaryEntry {
cluster_id: GlobalClusterId,
interval: Interval,
shard_id: ShardId,
}-
Dirty Key Collection: Each shard tracks keys modified since last reconciliation
-
Boundary Exchange: Router collects dirty boundaries from all shards
-
Merge Detection: For each key appearing on multiple shards:
fn detect_cross_shard_merges(key: &IdentityKeySignature, entries: &[BoundaryEntry]) -> Vec<ClusterMerge> { let mut merges = Vec::new(); // Group by overlapping intervals for (a, b) in entries.iter().tuple_combinations() { if a.shard_id != b.shard_id && a.interval.overlaps(&b.interval) { merges.push(ClusterMerge { primary: a.cluster_id, secondary: b.cluster_id, }); } } merges }
-
Merge Application: Each shard applies merges to its local DSU:
fn apply_cross_shard_merge(&mut self, primary: GlobalClusterId, secondary: GlobalClusterId) -> usize { // Update all records in secondary cluster to point to primary let mut updated = 0; for record_id in self.records_in_cluster(secondary) { self.cluster_map.insert(record_id, primary); updated += 1; } updated }
-
Key Clearing: Successfully reconciled keys are removed from dirty set
- Eventual Consistency: Cross-shard clusters converge after reconciliation
- No Data Loss: Failed reconciliation retries on next cycle
- Conflict Preservation: Cross-shard conflicts are detected and reported
For testing and small datasets:
struct Store {
records: FxHashMap<RecordId, Record>,
by_entity_type: FxHashMap<String, Vec<RecordId>>,
attr_interner: StringInterner,
value_interner: StringInterner,
}Production storage with column families:
| Column Family | Key | Value | Purpose |
|---|---|---|---|
records |
RecordId (4B) | Record (bincode) | Record storage |
index_identity |
hash (8B) | RecordId list | Identity key index |
index_attr_value |
attr:value | RecordId list | Attribute lookup |
dsu_parent |
RecordId | Parent RecordId | DSU parent links |
dsu_rank |
RecordId | u32 | DSU rank for balancing |
dsu_guards |
(RecordId, RecordId) | TemporalGuard | Merge guards |
cluster_assignments |
RecordId | ClusterId | Cluster membership |
interner |
String | InternedId | String interning |
metadata |
key | value | Manifest, counters |
[storage]
block_cache_mb = 512 # Read cache
write_buffer_mb = 128 # Write buffer before flush
max_background_jobs = 4 # Compaction threads
rate_limit_mbps = 0 # I/O rate limiting (0 = unlimited)-
Atomic DSU: Lock-free parent updates using CAS operations
fn find(&self, x: RecordId) -> RecordId { let mut current = x; loop { let parent = self.parent[current].load(Ordering::Acquire); if parent == current { return current; } // Path compression with CAS let grandparent = self.parent[parent].load(Ordering::Acquire); let _ = self.parent[current].compare_exchange( parent, grandparent, Ordering::Release, Ordering::Relaxed ); current = parent; } }
-
Sharded Caching: 256 shards to minimize contention
fn get_shard(&self, key: &K) -> &RwLock<LruCache<K, V>> { let hash = hash(key); &self.shards[hash as usize % 256] }
Identity key hashing uses SIMD for throughput:
fn simd_hash(data: &[u8]) -> u64 {
// Process 32 bytes at a time using AVX2
let mut state = _mm256_set1_epi64x(SEED);
for chunk in data.chunks_exact(32) {
let block = _mm256_loadu_si256(chunk.as_ptr() as *const _);
state = _mm256_xor_si256(state, block);
state = _mm256_mul_epi32(state, MULTIPLIER);
}
// Horizontal reduction
reduce_256_to_64(state)
}Write-ahead logging with coalescing:
struct AsyncWal {
tx: Sender<WalEntry>, // Submit writes
writer_thread: JoinHandle, // Background writer
}
impl AsyncWal {
fn submit(&self, data: Vec<u8>) -> WalTicket {
let ticket = WalTicket::new();
self.tx.send(WalEntry { data, ticket: ticket.clone() });
ticket // Caller can wait on ticket
}
}
// Writer thread coalesces multiple writes into single fsync
fn wal_writer_loop(rx: Receiver<WalEntry>, config: WalConfig) {
let mut buffer = Vec::new();
loop {
match rx.recv_timeout(config.max_coalesce_delay) {
Ok(entry) => buffer.push(entry),
Err(Timeout) => {
if !buffer.is_empty() {
flush_buffer(&mut buffer); // Single fsync
}
}
}
if buffer.len() >= config.max_coalesce_records {
flush_buffer(&mut buffer);
}
}
}For non-persistent shards, records can be partitioned by identity key hash for
parallel processing. Each partition uses process_batch_optimized():
┌──────────────────────┐
│ Incoming Records │
└──────────┬───────────┘
│
┌──────────▼───────────┐
│ Partition by Hash │ (identity_key_hash % partition_count)
└──────────┬───────────┘
│
┌────────────────────┼────────────────────┐
│ │ │
┌──────▼──────┐ ┌──────▼──────┐ ┌──────▼──────┐
│ Partition 0 │ │ Partition 1 │ │ Partition N │
│ │ │ │ │ │
│ 1. Batch │ │ 1. Batch │ │ 1. Batch │
│ Insert │ │ Insert │ │ Insert │
│ 2. Parallel │ │ 2. Parallel │ │ 2. Parallel │
│ Extract │ │ Extract │ │ Extract │
│ 3. Seq Link │ │ 3. Seq Link │ │ 3. Seq Link │
└──────┬──────┘ └──────┬──────┘ └──────┬──────┘
│ │ │
└────────────────────┼────────────────────┘
│
┌──────────▼───────────┐
│ Merge & Sort by │
│ Original Index │
└──────────────────────┘
Each partition runs independently with its own Mutex<Partition>. Rayon processes all partitions in parallel—no global lock contention.
Partition stores are currently in-memory. A shard configured with data_dir
therefore does not use this path: it routes every batch through the shard's
PersistentStore, where records, indexes, and cluster assignments are durable
and immediately available to queries. Enabling durable partition-local stores
requires an explicit on-disk layout and recovery protocol; routing persistent
traffic to in-memory partitions is not a valid optimization.
1. Client sends RecordInput batch via gRPC
2. Router hashes identity keys → partition records to shards
3. Each shard receives its partition of the batch
4. Non-persistent shards may route large batches to ParallelPartitionedUnirust
5. On that path, records are partitioned again by identity_key_hash % partition_count
6. Each partition (in parallel via Rayon):
a. Batch insert all records to store
b. Parallel extract: identity keys + strong ID summaries (Rayon)
c. Sequential link: DSU merges with temporal guards
d. Update identity index
7. Persistent shards instead resolve through Unirust backed by PersistentStore
8. Before resolution, write and fsync a versioned, checksummed binary ingest WAL
9. Persist records, indexes, cluster assignments, and request metadata
10. Sync the RocksDB WAL to stable storage
11. Remove the ingest WAL and fsync its parent directory
12. Update boundary indexes for cross-shard tracking
13. Return cluster assignments
If a shard stops before step 11, restart replays the ingest WAL using source identity idempotency. If framing, length, or checksum validation fails, startup fails closed and preserves the corrupt file for operator recovery. Cross-shard merge redirects use the linker metadata column family and receive the same stable-storage barrier before their RPC reports success.
1. Client sends QueryEntitiesRequest
2. Router fans out to all shards (parallel)
3. Each shard:
a. Looks up descriptors in attribute index
b. Filters by temporal interval
c. Resolves cluster IDs via DSU
d. Computes golden records (conflict-free values)
4. Router aggregates results
5. Returns QueryOutcome:
- If single cluster matches: QueryMatches
- If multiple clusters claim same identity: QueryConflict
1. Router triggers reconciliation (periodic or on-demand)
2. Collect dirty boundary keys from all shards
3. For each key appearing on multiple shards:
a. Check for overlapping intervals
b. If overlap found: create merge candidates
4. Validate merges against temporal guards
5. Apply merges to each shard
6. Clear dirty keys
7. Report reconciliation stats
| Profile | Candidate Cap | Hot Key Threshold | Use Case |
|---|---|---|---|
| Balanced | 2,000 | 50,000 | General purpose |
| LowLatency | 1,000 | 20,000 | Fast responses |
| HighThroughput | 4,000 | 100,000 | Batch processing |
| BulkIngest | 500 | 10,000 | Large loads; full resolution with lower candidate caps |
| MemorySaver | 500 | 5,000 | Reduced memory |
| BillionScale | 2,000 + persistent DSU | 100,000 | Huge datasets |
The release audit on 2026-07-22 measured 50,598 records/sec for 10,000,000
records, 16 streams, 5,000-record batches, and 10% overlap on an Apple M5 with
32 GB RAM. All 10,000,000 records were acknowledged with zero stream errors.
The measurement includes an fsync of the RocksDB WAL before each successful
batch acknowledgement.
Earlier 280K-500K figures measured an in-memory partition path that did not persist or expose its records through the shard's primary store. They are not valid production baselines. Performance work must preserve full entity resolution and acknowledge records only after durable storage.
| Component | Memory per Million Records |
|---|---|
| In-memory DSU | ~12 MB |
| Persistent DSU | ~4 MB (cached) |
| Identity Index | ~50 MB |
| Record Storage | ~100 MB (compressed) |
| Operation | P50 | P99 |
|---|---|---|
| Single record ingest | 0.5ms | 2ms |
| Batch ingest (1000 records) | 10ms | 50ms |
| Point query | 0.2ms | 1ms |
| Range query (1000 results) | 5ms | 20ms |