Streaming Matrix & Graph Lab

Frame the Matrix and Graph Questions

Imagine watching a city through two narrow control-room displays. One display receives measurement rows—traffic, temperature, vibration—and must preserve the dominant directions without storing every row. The other receives road openings and closures and must answer whether two districts remain connected without retaining an expensive full history. Both problems compress a changing stream, but the questions and guarantees are different.

For a tiny matrix stream, rows (1,0), (0,1), (1,0) give the Gram matrix AᵀA=[[2,0],[0,1]]; the first coordinate is the dominant direction. For a graph stream, insert edges (A,B) and (B,C): A reaches C. Delete (B,C): A no longer reaches C. A low-rank summary must approximate quadratic energy ||Ax||²; a connectivity structure must preserve or recover enough crossing edges to certify components. Neither answer substitutes for the other.

A row stream appends vectors; a turnstile stream adds and removes weighted items, such as edges. A subspace approximation preserves the action of a matrix on important directions. Frequent Directions deterministically shrinks singular values to keep a compact row sketch; a randomized embedding applies a linear map with a probabilistic norm guarantee. Sparse recovery reconstructs a sparse vector from fewer linear measurements. An ℓ₀-sampler returns a uniformly distributed nonzero coordinate under its contract. A graph sparsifier preserves selected cut or spectral properties, while connectivity asks only whether paths exist.

Stakeholders include ML/platform engineers monitoring feature drift, graph-system teams tracking network reachability, SREs, capacity planners, security reviewers, and researchers comparing sketches. Acceptance requires explicit stream semantics, bounded state, deterministic seeded replay, version-compatible merges, exact small-instance oracles, mathematical error evidence, safe deletion handling, recoverable checkpoints, visible degradation, and no claim that a shared linear API creates a shared guarantee.

Architect Shared Streaming Primitives

The laboratory serves two workloads. Matrix producers append dense or sparse rows with up to 10,000 features; graph producers submit strict-turnstile edge updates over up to 100 million vertices. The envelope is {tenant, stream, update_id, event_time, kind, dimension_or_vertex_schema, payload, weight, algorithm_version}. A gateway authenticates, validates dimensions and weights, canonicalizes undirected edges, deduplicates update IDs, and writes an ordered replay log. Partition workers update exact shadow state for small streams and bounded sketches for production scale.

The shared measurement layer owns deterministic seed derivation, signed hashing, vector accumulation, numeric overflow checks, compatible merge, checkpoint epochs, and checksums. The matrix plane offers Frequent Directions for deterministic covariance-loss bounds and sparse/random embeddings for subspace tests. The graph plane maintains hierarchical sampled/recovery states over incidence/cut vectors and a spanning forest; replacement-edge recovery repairs deletions. Each result names stream model, algorithm, dimensions, seed ID, error/failure probability, epoch, watermark, and degradation status.

Targets include 250,000 updates/s per worker, update p99 under 5 ms, query p99 under 100 ms for bounded queries, checkpoints under 30 seconds, recovery under 10 minutes, and memory below 8 GiB per worker. Privacy limits isolate tenants and restrict raw rows/edges because feature directions and relationships can be sensitive. Threats include malformed sparse indices, NaN/overflow, adversarial cancellation, hash manipulation, deletion underflow, cardinality floods, replay duplication, corrupt checkpoints, cross-version merge, and unauthorized topology queries.

flowchart TD
    P[Row and edge producers] --> G[Auth validate canonicalize]
    G --> L[Partitioned replay log]
    L --> M[Shared measurement runtime]
    M --> F[Frequent Directions]
    M --> E[Random embeddings]
    M --> Z[Graph recovery levels]
    F --> Q[Matrix query API]
    E --> Q
    Z --> C[Connectivity and sparsifier API]
    M --> S[Versioned checkpoints]
    Q --> R[Result plus guarantee]
    C --> R
    O[Operators] --> K[Config rollout control]
    K --> M
    M --> T[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,L,M,K teal;
    class F,E,Z,S purple;
    class Q,C,R,T crimson;

Integrate Low-Rank and Connectivity Workflows

Build vertical slices. Slice 1 validates row/edge envelopes and maintains exact Gram, adjacency, and replay oracles. Slice 2 introduces the shared deterministic hashing and additive measurement interface. Slice 3 adds Frequent Directions with covariance tests. Slice 4 adds graph sampling/recovery levels and spanning-forest repair. Slice 5 adds partition merges, snapshots, migrations, shadow comparisons, and degraded fallbacks. Each slice can ingest, query, checkpoint, restore, and prove its resource ceiling independently.

Stable interfaces are apply(update), query(request), merge(other), snapshot(), and restore(bytes), with capability metadata separating append-only matrices from strict-turnstile graphs. Configuration validates feature dimension, sketch rows, vertex range, recovery levels, numeric type, seed identity, and cell/edge budgets before allocation. Snapshots contain algorithm/version, stream model, dimensions, seeds, epoch, watermark, checksum, and payload. Unsupported versions trigger replay or STATE_INCOMPATIBLE; safe fallback may serve a labelled exact small-state result, batch rebuild, stale checkpoint, or explicit rejection.

from dataclasses import dataclass
from collections import defaultdict, deque
from typing import Iterable

@dataclass(frozen=True)
class Config:
    dimension: int = 2
    max_vertices: int = 1000
    max_edges: int = 10_000
    format_version: int = 1
    def validate(self) -> None:
        if not 1 <= self.dimension <= 10_000: raise ValueError("unsafe matrix dimension")
        if not 1 <= self.max_vertices <= 1_000_000: raise ValueError("unsafe vertex limit")
        if not 1 <= self.max_edges <= 10_000_000: raise ValueError("unsafe edge limit")
        if self.format_version != 1: raise ValueError("unsupported format version")

class MatrixGraphLab:
    def __init__(self, config: Config):
        config.validate(); self.config = config
        self.gram = [[0.0]*config.dimension for _ in range(config.dimension)]
        self.rows = 0
        self.edges: dict[tuple[str,str], int] = {}

    def add_row(self, row: Iterable[float]) -> None:
        values = tuple(float(x) for x in row)
        if len(values) != self.config.dimension: raise ValueError("row dimension mismatch")
        if any(x != x or abs(x) > 1e100 for x in values): raise ValueError("non-finite/unsafe value")
        for i, x in enumerate(values):
            for j, y in enumerate(values): self.gram[i][j] += x*y
        self.rows += 1

    def update_edge(self, u: str, v: str, delta: int) -> None:
        if not u or not v or u == v or delta not in (-1,1): raise ValueError("invalid edge update")
        edge = tuple(sorted((u,v)))
        current = self.edges.get(edge, 0)
        if current + delta not in (0,1): raise ValueError("strict-turnstile edge violation")
        if delta == 1 and current == 0 and len(self.edges) >= self.config.max_edges:
            raise MemoryError("edge budget exceeded")
        if current + delta == 0: self.edges.pop(edge, None)
        else: self.edges[edge] = 1

    def energy(self, direction: Iterable[float]) -> float:
        x = tuple(float(v) for v in direction)
        if len(x) != self.config.dimension: raise ValueError("direction mismatch")
        return sum(x[i]*self.gram[i][j]*x[j]
                   for i in range(self.config.dimension) for j in range(self.config.dimension))

    def connected(self, source: str, target: str) -> bool:
        if source == target: return True
        graph = defaultdict(list)
        for (u,v) in self.edges: graph[u].append(v); graph[v].append(u)
        queue, seen = deque([source]), {source}
        while queue:
            node = queue.popleft()
            for nxt in graph[node]:
                if nxt == target: return True
                if nxt not in seen: seen.add(nxt); queue.append(nxt)
        return False

    def merge(self, other: "MatrixGraphLab") -> "MatrixGraphLab":
        if self.config != other.config: raise ValueError("incompatible state")
        overlap = set(self.edges) & set(other.edges)
        if overlap: raise ValueError("edge partitions overlap")
        out = MatrixGraphLab(self.config); out.rows = self.rows + other.rows
        out.gram = [[self.gram[i][j] + other.gram[i][j] for j in range(self.config.dimension)]
                    for i in range(self.config.dimension)]
        out.edges = self.edges | other.edges
        if len(out.edges) > self.config.max_edges: raise MemoryError("merged edge budget exceeded")
        return out

def self_test() -> None:
    cfg = Config(dimension=2, max_vertices=10, max_edges=10)
    lab = MatrixGraphLab(cfg)
    for row in [(1,0),(0,1),(1,0)]: lab.add_row(row)
    assert lab.gram == [[2.0,0.0],[0.0,1.0]]
    assert lab.energy((1,0)) == 2 and lab.energy((0,1)) == 1
    lab.update_edge("A","B",1); lab.update_edge("B","C",1)
    assert lab.connected("A","C")
    lab.update_edge("B","C",-1); assert not lab.connected("A","C")
    left, right = MatrixGraphLab(cfg), MatrixGraphLab(cfg)
    left.add_row((1,0)); left.update_edge("A","B",1)
    right.add_row((0,1)); right.update_edge("C","D",1)
    merged = left.merge(right)
    assert merged.gram == [[1.0,0.0],[0.0,1.0]] and len(merged.edges) == 2
    try: lab.update_edge("B","C",-1)
    except ValueError: pass
    else: raise AssertionError("deletion underflow accepted")
    try: left.merge(MatrixGraphLab(Config(dimension=3)))
    except ValueError: pass
    else: raise AssertionError("incompatible merge accepted")
    print("integration checks passed", lab.gram, lab.connected("A","C"))

if __name__ == "__main__": self_test()

flowchart TD
    C[Typed update contracts] --> X[Exact Gram and graph oracles]
    X --> M[Shared linear measurements]
    M --> F[Frequent Directions slice]
    M --> L[Graph recovery levels]
    F --> U[Unified versioned state]
    L --> U
    U --> S[Snapshot merge replay]
    S --> K[Shadow canary]
    K -->|Oracle and SLO pass| P[Progressive promotion]
    K -->|Error or invariant fail| R[Rollback exact baseline]
    R --> X
    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,X navy;
    class M,F,L,U teal;
    class S,K purple;
    class P,R crimson;

Validate Accuracy, Scale and Recovery

Verification begins with exact algebra and graph traversal. Unit tests cover outer products, deterministic hashing, signed updates, shrinkage, sampling levels, forest operations, serialization, and compatibility. Property tests assert Gram symmetry and positive semidefinite quadratic forms, Frequent Directions covariance-loss bounds on small matrices, embedding norm preservation across preregistered seeds, strict-turnstile nonnegativity, forest acyclicity, and equality between connectivity answers and BFS on active-edge oracles.

An approachable end-to-end case is the two-dimensional matrix and three-vertex graph above. The professional workload combines 10,000-feature sparse rows with rank/drift changes and a 100-million-vertex graph with bursty inserts, adversarial deletions, bridge-edge churn, dense communities, and partition merges. Targets include 250,000 updates/s/worker, update p99 <5 ms, bounded query p99 <100 ms, matrix error within configured theorem/empirical envelope, zero connectivity mismatch on oracle-sized shadows, memory <8 GiB, checkpoint <30 s, and recovery <10 min.

Fault injection kills a worker during checkpoint, drops a partition, corrupts a measurement cell, rotates a seed without a version bump, introduces NaN/overflow, reorders deletions, exhausts one recovery level, and delays storage. Metrics expose lag, rejected/duplicate updates, numeric magnitude, shrink events, residual/covariance error, sampler failures, recovery attempts, forest repairs, shadow BFS mismatch, state bytes, checkpoint age, and incompatible merges. Traces connect update, measurement, query, recovery, and checkpoint; audit logs preserve configuration, sensitive queries, changes, incidents, and decisions.

Demonstrate the Matrix and Graph Lab

The release bundle contains update/query schemas, stream-model definitions, matrix and graph adapter contracts, parameter calculators, pinned dependencies, seeded datasets, benchmark commands and raw results, mathematical/correctness dossier, migration matrix, privacy review, dashboards, alerts, and an operational runbook. Every artifact names its version, owner, generation command, review date, and the theorem or empirical claim it supports.

The runbook begins with numeric growth, covariance drift, sampler/recovery failure, forest mismatch, deletion underflow, lag, quota breach, stale checkpoint, and incompatible state. It gives diagnostics, thresholds, safe fallbacks, escalation, rollback, and evidence preservation. The demonstration computes the tiny Gram matrix and connectivity trace, shows separate guarantee envelopes, partitions and merges compatible state, rejects dimension/seed mismatch, restores a checkpoint, corrupts one cell, and ends with the professional workload dashboard.

Peer reviewers ask: Are append-only and turnstile semantics explicit? Which operations are genuinely linear and mergeable? Does every query cite the right norm or failure probability? Can adversarial cancellation or overflow invalidate state? What happens when recovery fails? Can a matrix projection or graph sketch reveal sensitive relationships? Handoff requires tagged source, checksums, one-command tests, benchmark provenance, signed dossier, rehearsed recovery/rollback, on-call ownership, and no undocumented approximation.