Similarity & Deduplication Service

Define Duplicate, Similar and Acceptably Different

A similarity and deduplication service is like a librarian deciding whether a newly delivered book is already in the catalogue. An identical barcode is easy; a revised cover, corrected spelling, reordered paragraph, or resized image requires evidence. The service turns each item into comparable features, retrieves a small candidate set, verifies candidates with a declared similarity rule, and records a decision rather than comparing every new item with everything ever seen.

Take two token sets: A={red, fox, runs, fast} and B={red, fox, runs, today}. Their intersection has 3 elements and their union has 5, so Jaccard similarity is 3/5 = 0.60. If policy says “near duplicate at 0.80 or above,” they are acceptably different even though an approximate signature might place them in the same candidate bucket. Conversely, failing to generate a true candidate is a false negative that exact reranking cannot repair.

A shingle is a contiguous token or character sequence, such as word bigram red fox. A set signature such as MinHash estimates Jaccard similarity; a vector signature such as SimHash approximates angular similarity. Locality-sensitive hashing (LSH) groups likely neighbours into shared buckets. A membership filter can cheaply say “definitely unseen” or “possibly seen” under its contract. Verification computes the authoritative business metric over retained canonical features. The index lifecycle covers insert, tombstone, compaction, rebuild, migration, and rollback.

Stakeholders include content owners, storage engineers, search/recommendation teams, abuse analysts, privacy reviewers, and operators. Acceptance requires versioned normalization and signatures, tenant isolation, bounded candidate counts and memory, deterministic replay, measured recall at the decision threshold, exact verification before destructive action, explainable outcomes, and reversible merges or quarantines.

Design the Signature and Candidate Pipeline

The service accepts up to 20,000 documents per second across isolated tenants, with 10 million active items per large tenant, candidate lookup below 50 ms p99, final decision below 200 ms p99, and a fixed per-item index budget. Producers send {tenant_id, item_id, content_uri, media_type, source_time, normalization_version, idempotency_key}. The gateway authenticates, validates size/type, resolves governed content, and writes a replayable command. Content is normalized, shingled, signed, optionally screened by compact filters, inserted into versioned LSH bands, and verified against a capped candidate set.

For text, word or character shingles feed 128-row MinHash; 32 bands of 4 rows give candidate probability 1-(1-s^4)^32 at true Jaccard similarity s. Parameters are calibrated from a labelled corpus rather than selected by folklore. SimHash is available for weighted bag-of-feature cosine-like workloads, but not mixed silently with MinHash. Exact Jaccard verification uses retained canonical shingle sets or a governed feature store. Every result reports candidate method/version, exact score, threshold/policy version, explanation, and decision state.

Privacy limits prohibit cross-tenant candidate generation, restrict raw-content access, encrypt features and snapshots, minimize retained shingles, and audit reviews. The threat model includes adversarial token stuffing, hash flooding, oversized content, crafted bucket explosions, duplicate command replay, malicious feedback, poisoned thresholds, snapshot tampering, and operator credential misuse. Resource guards cap shingles, candidates, buckets per item, verification bytes, and per-tenant index growth before allocation.

flowchart TD
    P[Content producers] --> G[Auth validate quota]
    G --> N[Normalize and shingle]
    N --> S[MinHash or SimHash]
    S --> F[Compact membership screen]
    F --> L[Versioned LSH index]
    L --> C[Bounded candidates]
    C --> V[Exact verifier]
    V --> D[Keep link or review]
    D --> X[Index and decision log]
    O[Operators] --> R[Policies rollout feedback]
    R --> V
    X --> M[Metrics traces audit]
    classDef navy fill:#17365D,color:#fff,stroke:#102844;
    classDef teal fill:#147D84,color:#fff,stroke:#0c5c61;
    classDef crimson fill:#C6284D,color:#fff,stroke:#8e1935;
    classDef purple fill:#6C3BAA,color:#fff,stroke:#482675;
    class P,O navy;
    class G,N,S,F teal;
    class L,C,X purple;
    class V,D,R,M crimson;

Build Indexing, Verification and Feedback Loops

Build vertical slices. Slice 1 normalizes text, calculates an exact content hash, and compares every small-corpus item with exact Jaccard. Slice 2 adds deterministic MinHash and an LSH candidate index while retaining the oracle. Slice 3 introduces compact membership filters, partitioning, snapshots, tombstones, and rebuilds. Slice 4 adds policy decisions, explanations, reviewer feedback, and a shadow index for migrations. Each slice demonstrates ingest, query, verification, deletion, restore, and rollback independently.

Stable interfaces are upsert(item), candidates(item, limit), verify(left,right,policy), decide(item), tombstone(id,epoch), snapshot(), and restore(bytes). Configuration validates shingle size, signature length, band divisibility, threshold range, maximum shingles/candidates/bucket size, seeds, and format version before allocating. Serialized state records normalization, hash, signature, LSH, and policy versions plus tenant, epoch, checksum, and encryption metadata. Unsupported migrations rebuild from canonical features rather than coercing state.

Feedback is evidence, not truth by fiat. Reviewer labels are immutable events with actor, rationale, policy, and source versions. They feed threshold evaluation and labelled regression sets only after quality checks. Safe degradation can run exact-hash-only mode, quarantine a result, or reject over-budget content. It must not silently lower the threshold, cross tenant boundaries, or convert “no candidate found” into proof of uniqueness.

from dataclasses import dataclass
from hashlib import blake2b
from collections import defaultdict
import re

def stable_hash(seed: int, value: str) -> int:
    return int.from_bytes(blake2b(f"{seed}:{value}".encode(), digest_size=8).digest(), "big")

def shingles(text: str, width: int) -> frozenset[str]:
    tokens = re.findall(r"[a-z0-9]+", text.lower())
    return frozenset(" ".join(tokens[i:i+width]) for i in range(max(0, len(tokens)-width+1)))

def jaccard(a: frozenset[str], b: frozenset[str]) -> float:
    return len(a & b) / len(a | b) if a or b else 1.0

@dataclass(frozen=True)
class Config:
    shingle_width: int = 2
    rows: int = 32
    bands: int = 8
    threshold: float = 0.75
    max_shingles: int = 10_000
    max_candidates: int = 100
    seed: int = 17
    format_version: int = 1

    def validate(self) -> None:
        if not 1 <= self.shingle_width <= 8:
            raise ValueError("invalid shingle width")
        if self.rows <= 0 or self.bands <= 0 or self.rows % self.bands:
            raise ValueError("rows must divide evenly into bands")
        if not 0.0 <= self.threshold <= 1.0:
            raise ValueError("invalid threshold")
        if not 1 <= self.max_candidates <= 10_000 or not 1 <= self.max_shingles <= 1_000_000:
            raise ValueError("unsafe resource limit")
        if self.format_version != 1:
            raise ValueError("unsupported format version")

class Deduplicator:
    def __init__(self, config: Config):
        config.validate()
        self.config = config
        self.features: dict[str, frozenset[str]] = {}
        self.signatures: dict[str, tuple[int, ...]] = {}
        self.index: dict[tuple[int, tuple[int, ...]], set[str]] = defaultdict(set)

    def signature(self, features: frozenset[str]) -> tuple[int, ...]:
        if not features:
            return tuple(0 for _ in range(self.config.rows))
        return tuple(min(stable_hash(self.config.seed + row, token) for token in features)
                     for row in range(self.config.rows))

    def _band_keys(self, signature: tuple[int, ...]):
        size = self.config.rows // self.config.bands
        for band in range(self.config.bands):
            yield band, signature[band*size:(band+1)*size]

    def upsert(self, item_id: str, text: str) -> list[dict]:
        if not item_id or item_id in self.features:
            raise ValueError("item id must be new and non-empty")
        features = shingles(text, self.config.shingle_width)
        if len(features) > self.config.max_shingles:
            raise MemoryError("shingle budget exceeded")
        sig = self.signature(features)
        candidate_ids = set()
        for key in self._band_keys(sig):
            candidate_ids.update(self.index[key])
        if len(candidate_ids) > self.config.max_candidates:
            raise RuntimeError("candidate budget exceeded; quarantine item")
        results = [{"candidate": cid, "score": jaccard(features, self.features[cid])}
                   for cid in sorted(candidate_ids)]
        results.sort(key=lambda x: (-x["score"], x["candidate"]))
        self.features[item_id], self.signatures[item_id] = features, sig
        for key in self._band_keys(sig):
            self.index[key].add(item_id)
        return [{**r, "duplicate": r["score"] >= self.config.threshold,
                 "policy_threshold": self.config.threshold,
                 "algorithm": "minhash_lsh_then_exact_jaccard"} for r in results]

def self_test() -> None:
    cfg = Config(threshold=0.75, rows=32, bands=8, seed=19)
    service = Deduplicator(cfg)
    assert service.upsert("A", "red fox runs very fast") == []
    result = service.upsert("B", "red fox runs very fast today")
    assert result and result[0]["candidate"] == "A"
    assert result[0]["score"] == 4/5 and result[0]["duplicate"] is True
    assert service.upsert("C", "database backup completed nightly") == []
    assert service.signature(service.features["A"]) == Deduplicator(cfg).signature(service.features["A"])
    try:
        Deduplicator(Config(rows=31, bands=8))
    except ValueError:
        pass
    else:
        raise AssertionError("incompatible band layout accepted")
    try:
        service.upsert("A", "replacement")
    except ValueError:
        pass
    else:
        raise AssertionError("non-idempotent overwrite accepted")
    print("integration checks passed", result)

if __name__ == "__main__":
    self_test()

flowchart TD
    C[Contracts and exact hash] --> O[All-pairs Jaccard oracle]
    O --> S[Deterministic signatures]
    S --> L[LSH candidate index]
    L --> V[Exact policy verifier]
    V --> I[Lifecycle and snapshots]
    I --> F[Reviewed feedback loop]
    F --> K[Shadow index canary]
    K -->|Recall and SLO pass| P[Promote version]
    K -->|Regression or drift| R[Rollback index]
    R --> O
    classDef navy fill:#17365D,color:#fff,stroke:#102844;
    classDef teal fill:#147D84,color:#fff,stroke:#0c5c61;
    classDef crimson fill:#C6284D,color:#fff,stroke:#8e1935;
    classDef purple fill:#6C3BAA,color:#fff,stroke:#482675;
    class C,O navy;
    class S,L,V,I teal;
    class F,K purple;
    class P,R crimson;

Measure Recall, Cost and Adversarial Failures

Verification treats candidate recall and decision precision as separate contracts. Unit tests cover normalization, shingle boundaries, deterministic hashing, band slicing, exact scoring, idempotency, tombstones, and serialization. Property tests assert Jaccard symmetry and bounds, identical inputs score 1, signature equality estimates set similarity across seeds, compatible rebuilds reproduce postings, and tenant IDs never share buckets. Differential replay compares LSH candidates and final decisions with all-pairs exact oracles on manageable corpora.

The approachable case uses A, B, and C from the build exercise. The professional workload contains 10 million items per large tenant, 20,000 inserts/s, multilingual text, templated pages, OCR noise, boilerplate, adversarial token padding, and labelled duplicate clusters near the threshold. Targets are candidate recall ≥99.5% for exact Jaccard ≥0.85, verified-decision precision ≥99.9%, candidate p99 <50 ms, end-to-end p99 <200 ms, availability ≥99.9%, candidate cap ≤100, and a published cost per million items.

Fault injection kills an indexer during checkpoint, delays a partition, corrupts postings, rotates a seed without a version bump, fills one hot bucket, removes the feature store, replays commands, and poisons feedback labels. Metrics expose ingest lag, bucket size, candidates/query, cap rejections, verification score distribution, labelled recall/precision, shadow-index disagreement, tombstone lag, snapshot age, and cross-version attempts. Traces connect normalization, signature, bands, candidate reads, exact verification, and decision. Audit evidence preserves policy/version, reviewer action, explanation, rollback, and content-access purpose.

Deliver the Deduplication Service

The release bundle contains API and event schemas, normalization and feature specifications, signature/LSH parameter calculators, policy examples, a pinned environment, labelled datasets and lineage, benchmark commands and raw results, the correctness dossier, migration matrix, privacy/security review, dashboards, alerts, and an operational runbook. Every artifact states version, owner, generation command, review date, and retention rule.

The runbook begins with symptoms—candidate explosion, recall drift, latency, index disagreement, tombstone lag, feature-store outage, mixed versions—and maps each to diagnostics, thresholds, safe degradation, escalation, rollback, and evidence preservation. The demonstration normalizes A/B/C, shows exact hashes and shingles, retrieves B as A's candidate, verifies 4/5, keeps C distinct, rejects an incompatible layout, tombstones and rebuilds an item, and finishes with labelled recall/cost dashboards.

Peer reviewers ask: Which differences does normalization erase? Can candidate recall be reproduced by cohort? Is every destructive outcome exactly verified and reversible? What happens when canonical features are unavailable? Can a hot bucket cross the resource budget? Are feedback labels governed? Could signatures reveal sensitive content? Handoff requires a tagged build, checksums, one-command verification, benchmark provenance, signed correctness/privacy dossiers, rehearsed rebuild/rollback, on-call ownership, and no undocumented bypass.