Approach (brief): Use DynamoDB as a deduplication (idempotency) store keyed by an idempotency token (or event id). Write with a conditional Put (attribute_not_exists) so only the first writer wins; store processing state/result and an expiry TTL. Handle hot keys by sharding the partition key, use exponential backoff with jitter on throttling, and read-after-write when necessary for at-least-once semantics.Pseudocode (Python-style):python
import time, random
from dynamodb_client import Table # pseudo client
TABLE = Table("dedup")
TTL_SECONDS = 24*3600 # keep dedup records 24h
SHARD_COUNT = 8 # adjust based on traffic per key
def shard_key(base_id):
# simple deterministic sharding to spread hot keys
shard = hash(base_id) % SHARD_COUNT
return f"{base_id}#{shard}"
def now_ts():
return int(time.time())
def put_if_not_exists(idempotency_token, payload):
pk = shard_key(idempotency_token)
item = {
"pk": pk,
"token": idempotency_token,
"state": "processing",
"payload": payload, # optional small metadata
"created_at": now_ts(),
"ttl": now_ts() + TTL_SECONDS
}
try:
# Conditional write: only succeed if token attribute not present
TABLE.put_item(item, condition_expression="attribute_not_exists(token)")
return {"created": True, "item": item}
except ConditionalCheckFailedException:
existing = TABLE.get_item({"pk": pk, "token": idempotency_token})
return {"created": False, "item": existing}
def mark_complete(idempotency_token, result):
pk = shard_key(idempotency_token)
# idempotent update: set state and result, conditional on state != 'done'
for attempt in retry_with_backoff():
try:
TABLE.update_item(
key={"pk": pk, "token": idempotency_token},
update_expression="SET #s = :done, #r = :res, ttl = :ttl",
expression_attribute_names={"#s":"state","#r":"result"},
expression_attribute_values={
":done":"done", ":res": result, ":ttl": now_ts()+TTL_SECONDS
},
condition_expression="attribute_exists(token) AND #s <> :done"
)
return True
except ConditionalCheckFailedException:
return False
except ProvisionedThroughputExceededException:
sleep = backoff_with_jitter(attempt)
time.sleep(sleep)
raise Exception("Throttled")
def ingestion_handler(event):
token = event["idempotency_token"]
res = put_if_not_exists(token, event.get("meta"))
if res["created"]:
# process the event (may be long-running)
result = process_event(event)
mark_complete(token, result)
return result
else:
# duplicate: return stored result if done, or wait/read until done
item = res["item"]
if item and item.get("state") == "done":
return item["result"]
# wait-loop with bounded backoff to achieve at-least-once returning stored outcome
for t in bounded_waits():
item = TABLE.get_item({"pk": shard_key(token), "token": token})
if item and item.get("state") == "done":
return item["result"]
time.sleep(t)
# fallback: reprocess or return duplicate-accepted response
return process_event(event) # safe if idempotent or accept duplicate side-effects
