compression.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Object compression and delta encoding for MWP wire transfer. |
| 2 | |
| 3 | Tier 1 — zlib: every object is zlib-compressed before transmission. |
| 4 | Tier 2 — delta+zlib: objects with a known base are delta-encoded, then zlib-compressed. |
| 5 | |
| 6 | Both tiers use only Python stdlib (zlib, struct) — no external dependencies. |
| 7 | The server always decompresses/reconstructs; the client always compresses. |
| 8 | """ |
| 9 | |
| 10 | from __future__ import annotations |
| 11 | |
| 12 | import struct |
| 13 | import zlib |
| 14 | |
| 15 | # Minimum byte-run length to emit a COPY instruction rather than DATA. |
| 16 | # 32 bytes covers most source-code line edits without excessive overhead. |
| 17 | _COPY_THRESHOLD = 32 |
| 18 | |
| 19 | |
| 20 | def compress_zlib(data: bytes) -> bytes: |
| 21 | """Compress *data* with zlib at level 1 (speed-optimised). |
| 22 | |
| 23 | Level 1 gives ~60 % reduction on source text with negligible CPU overhead |
| 24 | at the throughput rates of a pack push (hundreds of objects per second). |
| 25 | """ |
| 26 | return zlib.compress(data, level=1) |
| 27 | |
| 28 | |
| 29 | def decompress_zlib(data: bytes) -> bytes: |
| 30 | """Decompress zlib-compressed *data*. |
| 31 | |
| 32 | Raises :class:`zlib.error` on corrupt input — let it propagate so the |
| 33 | server returns 400 to the client rather than storing garbage. |
| 34 | """ |
| 35 | return zlib.decompress(data) |
| 36 | |
| 37 | |
| 38 | def compute_delta(base: bytes, target: bytes) -> bytes: |
| 39 | """Delta-encode *target* relative to *base*. |
| 40 | |
| 41 | Returns a zlib-compressed instruction stream. Each instruction is one of: |
| 42 | |
| 43 | * ``COPY``: ``b'\\x00' + struct.pack('>II', offset, length)`` |
| 44 | — copy *length* bytes from *base* starting at *offset*. |
| 45 | * ``DATA``: ``b'\\x01' + struct.pack('>I', length) + data_bytes`` |
| 46 | — emit *length* literal bytes verbatim. |
| 47 | |
| 48 | Algorithm: |
| 49 | Build a hash table mapping every *_COPY_THRESHOLD*-byte chunk of |
| 50 | *base* to its last position. Walk *target* looking for matching |
| 51 | chunks; extend each match forward as far as possible, then emit a |
| 52 | single COPY instruction. Unmatched bytes accumulate in a DATA buffer |
| 53 | that is flushed whenever a COPY is emitted. |
| 54 | |
| 55 | Raises :class:`ValueError` when the compressed delta is not smaller than |
| 56 | a plain zlib-compressed copy of *target*; the caller should fall back to |
| 57 | Tier 1 in that case. |
| 58 | |
| 59 | Args: |
| 60 | base: Previous version of the object (already stored on the remote). |
| 61 | target: New version of the object to encode. |
| 62 | |
| 63 | Returns: |
| 64 | Zlib-compressed binary instruction stream. |
| 65 | |
| 66 | Raises: |
| 67 | ValueError: Delta is not smaller than zlib(*target*) — not profitable. |
| 68 | """ |
| 69 | # Degenerate case: empty base → single DATA instruction. |
| 70 | if not base: |
| 71 | raw = b"\x01" + struct.pack(">I", len(target)) + target |
| 72 | return zlib.compress(raw, level=1) |
| 73 | |
| 74 | # Build hash table: chunk → last position in base. |
| 75 | table: dict[bytes, int] = {} |
| 76 | for i in range(len(base) - _COPY_THRESHOLD + 1): |
| 77 | table[base[i : i + _COPY_THRESHOLD]] = i |
| 78 | |
| 79 | parts: list[bytes] = [] |
| 80 | data_buf = bytearray() |
| 81 | pos = 0 |
| 82 | |
| 83 | while pos < len(target): |
| 84 | # Try to find a matching chunk starting at *pos*. |
| 85 | if pos + _COPY_THRESHOLD <= len(target): |
| 86 | chunk = target[pos : pos + _COPY_THRESHOLD] |
| 87 | base_pos = table.get(chunk) |
| 88 | if base_pos is not None: |
| 89 | # Extend match forward as far as bytes agree. |
| 90 | length = _COPY_THRESHOLD |
| 91 | while ( |
| 92 | pos + length < len(target) |
| 93 | and base_pos + length < len(base) |
| 94 | and target[pos + length] == base[base_pos + length] |
| 95 | ): |
| 96 | length += 1 |
| 97 | |
| 98 | # Flush accumulated DATA before emitting COPY. |
| 99 | if data_buf: |
| 100 | parts.append( |
| 101 | b"\x01" + struct.pack(">I", len(data_buf)) + bytes(data_buf) |
| 102 | ) |
| 103 | data_buf = bytearray() |
| 104 | |
| 105 | parts.append(b"\x00" + struct.pack(">II", base_pos, length)) |
| 106 | pos += length |
| 107 | continue |
| 108 | |
| 109 | # No match — accumulate literal byte. |
| 110 | data_buf.append(target[pos]) |
| 111 | pos += 1 |
| 112 | |
| 113 | # Flush trailing DATA. |
| 114 | if data_buf: |
| 115 | parts.append(b"\x01" + struct.pack(">I", len(data_buf)) + bytes(data_buf)) |
| 116 | |
| 117 | delta = zlib.compress(b"".join(parts), level=1) |
| 118 | |
| 119 | # Reject unprofitable deltas so the caller can fall back to plain zlib. |
| 120 | plain = zlib.compress(target, level=1) |
| 121 | if len(delta) >= len(plain): |
| 122 | raise ValueError( |
| 123 | f"delta ({len(delta)} B) not smaller than zlib({len(plain)} B) — not profitable" |
| 124 | ) |
| 125 | |
| 126 | return delta |
| 127 | |
| 128 | |
| 129 | def apply_delta(base: bytes, delta_compressed: bytes) -> bytes: |
| 130 | """Reconstruct the target by applying *delta_compressed* to *base*. |
| 131 | |
| 132 | Inverse of :func:`compute_delta`. Parses the instruction stream produced |
| 133 | by that function and assembles the target byte-for-byte. |
| 134 | |
| 135 | Args: |
| 136 | base: The base object bytes (fetched from storage). |
| 137 | delta_compressed: Zlib-compressed instruction stream. |
| 138 | |
| 139 | Returns: |
| 140 | Reconstructed target bytes. |
| 141 | |
| 142 | Raises: |
| 143 | zlib.error: *delta_compressed* is corrupt. |
| 144 | ValueError: Unknown instruction type in stream. |
| 145 | struct.error: Truncated instruction stream. |
| 146 | """ |
| 147 | stream = zlib.decompress(delta_compressed) |
| 148 | result = bytearray() |
| 149 | pos = 0 |
| 150 | n = len(stream) |
| 151 | |
| 152 | while pos < n: |
| 153 | instr_type = stream[pos] |
| 154 | pos += 1 |
| 155 | |
| 156 | if instr_type == 0: # COPY |
| 157 | offset, length = struct.unpack_from(">II", stream, pos) |
| 158 | pos += 8 |
| 159 | result.extend(base[offset : offset + length]) |
| 160 | |
| 161 | elif instr_type == 1: # DATA |
| 162 | (length,) = struct.unpack_from(">I", stream, pos) |
| 163 | pos += 4 |
| 164 | result.extend(stream[pos : pos + length]) |
| 165 | pos += length |
| 166 | |
| 167 | else: |
| 168 | raise ValueError(f"unknown delta instruction type: {instr_type:#04x} at stream offset {pos - 1}") |
| 169 | |
| 170 | return bytes(result) |