Fix silent data-corruption and lock contention in Codec

[gmr]

Summary

Fixes 3 verified bugs in codecs.py that caused silent data corruption and lock contention.

Problem / Solution

• Codec.decode silently returns raw bytes when optional msgpack/bs4 dependency is missing #83 Codec.decode silently returned raw bytes when the optional umsgpack/bs4 dependency was missing → now raises DecodeError when the content type matches but the library is absent, matching the Avro pattern.
• Codec._compress silently no-ops for unrecognized content_encoding values #96 _compress silently no-op'd for unrecognized content_encoding, emitting an uncompressed body whose AMQP property claimed a compression scheme → now raises EncodeError. (Decode-side pass-through is intentionally retained: a non-compression content_encoding like utf-8 is legitimate on decode.)
• Avro schema loading: failed loads retried under the single global lock, no negative caching, 4xx retried with backoff #97 Avro HTTP schema loading serialized every load behind one global lock and retried deterministic 4xx with backoff → per-message-type asyncio.Lock, fail-fast on 4xx (5xx/network errors still retry).

A stateless _decode_body helper was extracted to keep decode under the mccabe limit after adding the missing-dependency guards (no behavior change).

Testing

Full suite green (218 tests) + ruff clean. New tests/test_codecs.py (10 tests).

Closes #83
Closes #96
Closes #97

Summary by CodeRabbit

• Bug Fixes

  • Improved schema caching so concurrent schema fetches for different message types run independently.
  • Avro schema HTTP loading now fails fast on client errors while still retrying transient server errors.
  • Unsupported compression settings now raise a clear encode error instead of passing through.
  • Streamlined decoding for common formats with consistent fallback behavior.

• Tests

  • Added unit tests for missing optional dependencies, compression validation/round-trips, and Avro schema HTTP behavior (fail-fast, retries, concurrency, and no lock leakage).

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 7a31c1c2-0615-4132-90f4-ca24b31122dc

📥 Commits

Reviewing files that changed from the base of the PR and between 8e84c39 and b041d60.

📒 Files selected for processing (2)

• rejected/codecs.py
• tests/test_codecs.py

🚧 Files skipped from review as they are similar to previous changes (2)

• tests/test_codecs.py
• rejected/codecs.py

---

📝 Walkthrough

Walkthrough

Codec decode and encode behavior now raises errors for missing optional decoders and unsupported compression, while Avro schema loading uses per-message-type locks and fails fast on HTTP 4xx responses. New tests cover these decode, encode, and schema-loading paths.

Changes

Codec Decode/Encode Fixes

Layer / File(s)	Summary
Decode dispatch and missing dependencies rejected/codecs.py, tests/test_codecs.py	Non-Avro body decoding moved into _decode_body; matching msgpack and bs4 content types now raise DecodeError when the optional dependency is missing, and tests cover msgpack, HTML, and XML cases.
Compression validation rejected/codecs.py, tests/test_codecs.py	_compress now raises EncodeError for unsupported content_encoding values, with tests for the error path, gzip round-trip, and bzip2 output.
Per-type schema locking and HTTP 4xx fast-fail rejected/codecs.py, tests/test_codecs.py	Avro schema caching uses per-message_type locks, failed loads clear their lock entry, and HTTP 4xx responses stop retrying immediately; tests cover 404 fast-fail, 503 retries, concurrent loads, and same-type deduplication.

Estimated code review effort: 3 (Moderate) | ~25 minutes

Sequence Diagram(s)

sequenceDiagram
  participant Consumer
  participant Codec
  participant PerTypeLock
  participant SchemaRegistry

  Consumer->>Codec: decode(message_type)
  Codec->>PerTypeLock: acquire lock for message_type
  alt schema cached
    PerTypeLock-->>Codec: return cached schema
  else not cached
    Codec->>SchemaRegistry: GET schema
    alt 4xx response
      SchemaRegistry-->>Codec: 4xx status
      Codec-->>Consumer: raise DecodeError
    else 5xx response
      SchemaRegistry-->>Codec: 5xx status
      Codec->>SchemaRegistry: retry GET
      SchemaRegistry-->>Codec: 200 schema
      Codec->>PerTypeLock: cache schema
    else 200 response
      SchemaRegistry-->>Codec: 200 schema
      Codec->>PerTypeLock: cache schema
    end
  end
  Codec-->>Consumer: return decoded message

Loading

Poem

A rabbit tuned the codec tree,
With locks per type, so all run free.
No silent bytes for missing tricks,
No bad compressions, no loose fixes.
Hop hop — the schema finds its key. 🐇

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	Title succinctly describes the main change: fixing codec corruption and Avro lock contention.
Linked Issues check	✅ Passed	Changes address #83, #96, and #97: missing optional-dependency errors, unsupported compression, and Avro lock/retry behavior.
Out of Scope Changes check	✅ Passed	No unrelated code changes are evident; the helper refactor and tests support the codec fixes.
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

---

_{Comment @coderabbitai help to get the list of available commands.}

[gmr]

@coderabbitai review

[coderabbitai]

✅ Action performed

Review finished.

Note: CodeRabbit is an incremental review system and does not re-review already reviewed commits. This command is applicable only when automatic reviews are paused.

[coderabbitai]

Actionable comments posted: 1

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@rejected/codecs.py`:
- Line 78: Clean up stale per-message-type schema locks in the codec loader,
because failed schema loads leave entries behind in _schema_locks without ever
populating _avro_schemas. Update the load path around the schema fetch/cache
logic in the codec class (the code that uses _schema_locks and _avro_schemas) so
that when a load fails for invalid, missing, or 4xx schema names, the
corresponding lock entry is removed after the attempt completes. Make sure the
cleanup happens for the failing message_type even when the exception is raised,
while preserving the lock for successful loads.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 6eefa5b9-df6f-4207-bf40-9ea3d0fc8b60

📥 Commits

Reviewing files that changed from the base of the PR and between 401dd78 and 8e84c39.

📒 Files selected for processing (2)

• rejected/codecs.py
• tests/test_codecs.py