Implement an Idempotent Versioned Database Update

Problem Statement

In distributed systems, network partitions and client retries often cause identical mutation requests to arrive multiple times. Simultaneously, multiple workers may attempt to update the same database entity concurrently.

Implement an in-memory service layer component named VersionedRecordService that manages updates to single records while guaranteeing idempotency (retried requests produce identical responses without re-applying side effects) and optimistic concurrency control (detecting and rejecting stale concurrent writes).

Requirements

1. Record Schema: Each record contains record_id: str, version: int (monotonically increasing integer starting at 1 on creation), data: dict, and updated_at: float.
2. Idempotency Tracking: Every mutation is accompanied by an idempotency_key: str. If a request with an already-processed idempotency_key arrives for that record:
   • If the incoming payload (expected_version and updates) matches the original request, return the cached result of the previous execution without incrementing the version or altering timestamps.
   • If the payload differs under the same key, reject the request as a conflict/invalid request.
3. Optimistic Locking: If the incoming expected_version does not match the record's current version, reject the update with a version conflict error.
4. Thread Safety: The operation must be safe under concurrent calls within the same process.

Class Interface

Implement the following class in Python:

from dataclasses import dataclass
from typing import Any, Dict, Optional, Tuple

@dataclass
class Record:
    record_id: str
    version: int
    data: Dict[str, Any]
    updated_at: float

class VersionedRecordService:
    def __init__(self):
        pass

    def put_initial(self, record_id: str, data: Dict[str, Any]) -> Record:
        """Seeds an initial record at version 1."""
        pass

    def update_record(
        self,
        record_id: str,
        expected_version: int,
        updates: Dict[str, Any],
        idempotency_key: str,
        current_time: float
    ) -> Tuple[bool, Optional[Record], str]:
        """
        Attempts to apply updates to record_id.

        Returns: (success: bool, record: Optional[Record], message: str)
        - On success: (True, updated_record, "OK")
        - On idempotent replay: (True, cached_record, "IDEMPOTENT_REPLAY")
        - On version mismatch: (False, current_record, "VERSION_MISMATCH")
        - On idempotency conflict (same key, different payload): (False, None, "IDEMPOTENCY_CONFLICT")
        - On record not found: (False, None, "NOT_FOUND")
        """
        pass

Example

service = VersionedRecordService()
rec = service.put_initial("user_1", {"name": "Alice", "role": "member"})
# rec.version is 1

# 1. Successful update
ok, rec, msg = service.update_record(
    "user_1", expected_version=1, updates={"role": "admin"}, idempotency_key="key-100", current_time=1000.0
)
# Output: (True, Record(record_id="user_1", version=2, data={"name": "Alice", "role": "admin"}, updated_at=1000.0), "OK")

# 2. Network retry with identical key and payload
ok, rec, msg = service.update_record(
    "user_1", expected_version=1, updates={"role": "admin"}, idempotency_key="key-100", current_time=1005.0
)
# Output: (True, Record(version=2, ...), "IDEMPOTENT_REPLAY") -> version is still 2, updated_at remains 1000.0

# 3. Concurrent write with stale version
ok, rec, msg = service.update_record(
    "user_1", expected_version=1, updates={"role": "editor"}, idempotency_key="key-200", current_time=1010.0
)
# Output: (False, Record(version=2, ...), "VERSION_MISMATCH")

Constraints

• record_id and idempotency_key are non-empty strings of length <= 128.
• updates contains string keys and serializable primitive values.
• Idempotency records should be scoped per record or globally; memory overhead should be bounded or tracked per request.