Skip to content

Phase 2: Fix WAL dump to use operation log instead of raw cache #29

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
akeenkarkare merged 11 commits into from
Nov 21, 2025
Merged

Phase 2: Fix WAL dump to use operation log instead of raw cache #29

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
05e1d38
Fix WAL dump to use operation log instead of raw cache
akeenkarkare Nov 20, 2025
9dccec7
Fix critical DELETE validation and INCREMENT replay bugs
akeenkarkare Nov 20, 2025
f5dcd79
Document DELETE validation and INCREMENT replay fixes in plan
akeenkarkare Nov 20, 2025
37d2235
Fix list type conversion to preserve nested types
akeenkarkare Nov 20, 2025
35b0a8b
Document list type conversion fix in plan
akeenkarkare Nov 20, 2025
a983600
Add cache debug endpoint and fix duel ID mismatch
akeenkarkare Nov 20, 2025
2b6ce84
Address CodeRabbit review feedback on admin endpoints and WAL dump
akeenkarkare Nov 20, 2025
8abe72b
Fix critical checkpoint skip and race condition bugs in WAL dump
akeenkarkare Nov 20, 2025
5fd690e
Fix INCREMENT double-apply bug by stopping on first failure
akeenkarkare Nov 20, 2025
1cb2a12
Add cache viewer tab to log viewer HTML
akeenkarkare Nov 20, 2025
3b36ff8
Delete CACHE_FIX_PLAN.md
akeenkarkare Nov 20, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
316 changes: 214 additions & 102 deletions scripts/fastapi/cache_dumper.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,40 @@ def chunks(lst: List, n: int):
yield lst[i:i + n]


def convert_value_to_dynamodb(value):
"""
Convert a single Python value to DynamoDB format (recursive helper)

Args:
value: Python value of any type

Returns:
DynamoDB formatted value with type key
"""
if isinstance(value, bool):
# IMPORTANT: Check bool before int (bool is subclass of int in Python)
return {'BOOL': value}
elif isinstance(value, str):
return {'S': value}
elif isinstance(value, int):
return {'N': str(value)}
elif isinstance(value, float):
return {'N': str(value)}
elif isinstance(value, dict):
return {'M': convert_to_dynamodb_format(value)}
elif isinstance(value, list):
# Recursively convert each item in the list
dynamodb_list = []
for item in value:
dynamodb_list.append(convert_value_to_dynamodb(item))
return {'L': dynamodb_list}
elif value is None:
return {'NULL': True}
else:
# Fallback: convert to string
return {'S': str(value)}


def convert_to_dynamodb_format(data: Dict) -> Dict:
"""
Convert Python dict to DynamoDB format
Expand All @@ -41,29 +75,7 @@ def convert_to_dynamodb_format(data: Dict) -> Dict:
dynamodb_item = {}

for key, value in data.items():
if isinstance(value, str):
dynamodb_item[key] = {'S': value}
elif isinstance(value, int):
dynamodb_item[key] = {'N': str(value)}
elif isinstance(value, float):
dynamodb_item[key] = {'N': str(value)}
elif isinstance(value, bool):
dynamodb_item[key] = {'BOOL': value}
elif isinstance(value, dict):
dynamodb_item[key] = {'M': convert_to_dynamodb_format(value)}
elif isinstance(value, list):
# Convert list to DynamoDB List
dynamodb_list = []
for item in value:
if isinstance(item, str):
dynamodb_list.append({'S': item})
elif isinstance(item, (int, float)):
dynamodb_list.append({'N': str(item)})
elif isinstance(item, dict):
dynamodb_list.append({'M': convert_to_dynamodb_format(item)})
dynamodb_item[key] = {'L': dynamodb_list}
elif value is None:
dynamodb_item[key] = {'NULL': True}
dynamodb_item[key] = convert_value_to_dynamodb(value)

return dynamodb_item

Expand Down Expand Up @@ -127,104 +139,204 @@ def batch_write_to_dynamodb(table_name: str, items: List[Dict]) -> Dict:

async def dump_cache_to_db() -> Dict:
"""
Dump all dirty cache entries to DynamoDB
Dump all dirty cache entries to DynamoDB by replaying WAL operations.

CRITICAL FIX: Instead of dumping raw cache entries (which have wrapped structures
like {"success": True, "data": [...]}), we now read from the WAL operation log
which has correctly structured operations for DynamoDB updates.

Returns:
Dictionary with dump statistics
"""
info("🗄️ Starting cache dump to DynamoDB...")
info("🗄️ Starting cache dump to DynamoDB via WAL operations...")

try:
# Get all dirty entries from cache
dirty_entries = cache_manager.get_dirty_entries()

if not dirty_entries:
info("✅ No dirty entries to dump")
return {"success": True, "entries": 0, "message": "No dirty entries"}

info(f"📦 Found {len(dirty_entries)} dirty entries to dump")

# Group entries by table
users_items = []
daily_items = []
duels_items = []
bounties_items = []

for entry in dirty_entries:
cache_type = entry.get('cache_type')
# Get checkpoint to avoid replaying already-applied entries
last_applied = wal_manager.get_last_applied_sequence()
info(f"📍 Checkpoint: last_applied_sequence = {last_applied}")

# Get WAL entries since last checkpoint (+ 1 to get next unapplied entry)
wal_entries = wal_manager.get_entries_since(last_applied + 1)

if not wal_entries:
info("✅ No new WAL entries to sync")
return {"success": True, "entries": 0, "message": "No new WAL entries"}

info(f"📦 Found {len(wal_entries)} new WAL entries to sync to DynamoDB")

# Track stats
total_synced = 0
total_failed = 0
errors = []

Comment on lines 140 to 171

@coderabbitai coderabbitai Bot Nov 20, 2025

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

🧩 Analysis chain

Overall WAL replay + checkpoint design looks correct and addresses prior data issues.

Using WAL entries (with last_applied_sequence checkpoints) instead of raw cache dumps fixes the original “wrapped cache shapes” bug and makes INCREMENT idempotent across retries. Per-entry stats and only clearing WAL/marking cache synced when total_failed == 0 are good safety improvements.

Once the unknown-op + concurrency edge cases above are addressed, this should be a much more robust foundation for cache persistence.

Also applies to: 288-314

🏁 Script executed:

#!/bin/bash
# Sanity-check how many callers trigger dump_cache_to_db and in what contexts (admin-only, background, etc.)
rg -n "dump_cache_to_db" -C3

Length of output: 4560

I need to inspect the full implementation of dump_cache_to_db() and the code at lines 288–314 to verify the edge cases mentioned in the review.

🏁 Script executed:

#!/bin/bash
# Get the full dump_cache_to_db function (starting at line 140) and lines 288-314
sed -n '140,250p' scripts/fastapi/cache_dumper.py

echo "---"
echo "Lines 288-314:"
echo "---"
sed -n '288,314p' scripts/fastapi/cache_dumper.py

Length of output: 5478

🏁 Script executed:

#!/bin/bash
# Get the full dump_cache_to_db function and check lines 288-314
sed -n '140,280p' scripts/fastapi/cache_dumper.py
echo "---"
echo "Lines 288-314:"
echo "---"
sed -n '288,314p' scripts/fastapi/cache_dumper.py

Length of output: 6742

I need to check the exception handler and verify concurrency protections in the WAL manager.

🏁 Script executed:

#!/bin/bash
# Check exception handler and look for unknown operation handling
sed -n '250,290p' scripts/fastapi/cache_dumper.py

echo "---"
echo "Check wal_manager for concurrency locks/mutexes"
rg -n "class WALManager|def get_entries_since|def set_last_applied|lock|Lock|mutex" scripts/fastapi/wal_manager.py -A2

Length of output: 2964

Address silent handling of unknown WAL operation types and add concurrency protection for non-scheduler callers.

The implementation leaves two unresolved edge cases from the review:

  1. Unknown operations silently fail: If a WAL entry has an operation type other than UPDATE, PUT, DELETE, or INCREMENT, the code validates it passes but then skips the DynamoDB call without explicit error handling or warning. The operation silently doesn't increment total_synced.

  2. Concurrency race on non-scheduler callers: While the scheduler job has max_instances=1, dump_cache_to_db() is also callable from:

    • bounties.py (admin endpoint, line 126)
    • main.py shutdown handler (line 92)
    • main.py manual trigger (line 180)

    Between wal_manager.get_last_applied_sequence() and wal_manager.get_entries_since(), concurrent calls can read the same checkpoint and fetch identical entries, causing duplicate DynamoDB operations and violating the idempotence assumption.

The WALManager itself has proper locking, but dump_cache_to_db() needs atomic checkpoint + fetch semantics and explicit handling for unknown operation types.

🤖 Prompt for AI Agents 
In scripts/fastapi/cache_dumper.py around lines 140-171, add two fixes: (1) make
checkpoint+fetch atomic by using a wal_manager-provided atomic fetch/reserve
method (e.g. wal_manager.fetch_and_reserve_entries(start_seq) /
wal_manager.get_and_lock_entries_since(start_seq)) or, if that API doesn't
exist, acquire a module-level asyncio.Lock (or wal_manager.lock()) around the
sequence read and get_entries_since call so concurrent callers cannot read the
same checkpoint and duplicate work; after reserving entries ensure you mark them
applied only after successful DynamoDB writes. (2) Add explicit handling for
unknown WAL operation types: log an error/warning with the entry sequence and op
type, increment total_failed, append a structured error object to errors, and
continue to next entry (do not silently skip). Ensure both changes update
returned stats (total_synced, total_failed, errors) accordingly.

# Process each WAL operation until first failure
# CRITICAL: We must stop on first failure to prevent double-applying later INCREMENTs
# Example: seq 1 ✅, seq 2 ❌, seq 3 INCREMENT ✅ (checkpoint=1)
# Next run: seq 2 fails again, seq 3 replayed → INCREMENT applied twice!
for entry in wal_entries:
operation = entry.get('operation')
table = entry.get('table')
key = entry.get('key')
data = entry.get('data')

if not data:
continue

# Convert to DynamoDB format
dynamodb_item = convert_to_dynamodb_format(data)
# Track sequence for checkpoint management
entry_sequence = entry.get('sequence', -1)

# Validate based on operation type
# DELETE operations don't require data, all others do
if operation == "DELETE":
if not all([operation, table, key]):
total_failed += 1
error_msg = f"Incomplete DELETE entry at sequence {entry_sequence}: {entry}"
warning(error_msg)
errors.append(error_msg)
# STOP processing to prevent replaying later entries (especially INCREMENTs)
break
else:
if not all([operation, table, key, data]):
total_failed += 1
error_msg = f"Incomplete {operation} entry at sequence {entry_sequence}: {entry}"
warning(error_msg)
errors.append(error_msg)
# STOP processing to prevent replaying later entries (especially INCREMENTs)
break

Comment thread
coderabbitai[bot] marked this conversation as resolved.
# CRITICAL FIX: Convert string cache_type to CacheType enum for comparison
# cache_type is a string (e.g., "users") from cache key, not a CacheType enum
try:
cache_type_enum = CacheType(cache_type)
except ValueError:
error(f"Unknown cache type: {cache_type}, skipping entry")
continue

# Route to appropriate table
if cache_type_enum == CacheType.USERS:
users_items.append(dynamodb_item)
elif cache_type_enum == CacheType.DAILY_PROBLEM or cache_type_enum == CacheType.DAILY_COMPLETIONS:
daily_items.append(dynamodb_item)
elif cache_type_enum == CacheType.DUELS:
duels_items.append(dynamodb_item)
elif cache_type_enum == CacheType.BOUNTIES or cache_type_enum == CacheType.BOUNTY_COMPETITIONS:
bounties_items.append(dynamodb_item)

# Batch write to each table
results = {}

if users_items:
info(f"💾 Writing {len(users_items)} users to DynamoDB...")
results['users'] = batch_write_to_dynamodb(USERS_TABLE, users_items)

if daily_items:
info(f"💾 Writing {len(daily_items)} daily items to DynamoDB...")
results['daily'] = batch_write_to_dynamodb(DAILY_TABLE, daily_items)

if duels_items:
info(f"💾 Writing {len(duels_items)} duels to DynamoDB...")
results['duels'] = batch_write_to_dynamodb(DUELS_TABLE, duels_items)

if bounties_items:
info(f"💾 Writing {len(bounties_items)} bounties to DynamoDB...")
results['bounties'] = batch_write_to_dynamodb(BOUNTIES_TABLE, bounties_items)

# Check overall success
all_successful = all(r.get('success', False) for r in results.values())

if all_successful:
# Mark all entries as synced
for entry in dirty_entries:
cache_manager.mark_synced(entry['cache_type'], entry.get('identifier', ''))

# Clear WAL file
wal_manager.clear()

total_written = sum(r.get('written', 0) for r in results.values())
info(f"✅ Cache dump complete: {total_written} entries written to DynamoDB")
if operation == "UPDATE":
# Build UpdateExpression from data
update_expr_parts = []
expr_attr_values = {}

for field, value in data.items():
update_expr_parts.append(f"{field} = :{field}")
# Convert to DynamoDB format using helper to preserve types
expr_attr_values[f":{field}"] = convert_value_to_dynamodb(value)

# Convert key to DynamoDB format
dynamodb_key = {}
for k, v in key.items():
if isinstance(v, str):
dynamodb_key[k] = {'S': v}
elif isinstance(v, (int, float)):
dynamodb_key[k] = {'N': str(v)}

# Perform update
ddb.update_item(
TableName=table,
Key=dynamodb_key,
UpdateExpression=f"SET {', '.join(update_expr_parts)}",
ExpressionAttributeValues=expr_attr_values
)

elif operation == "PUT":
# Full item put - merge key and data
item = {**key, **data}
dynamodb_item = convert_to_dynamodb_format(item)

ddb.put_item(
TableName=table,
Item=dynamodb_item
)

elif operation == "DELETE":
# Convert key to DynamoDB format
dynamodb_key = {}
for k, v in key.items():
if isinstance(v, str):
dynamodb_key[k] = {'S': v}
elif isinstance(v, (int, float)):
dynamodb_key[k] = {'N': str(v)}

ddb.delete_item(
TableName=table,
Key=dynamodb_key
)

elif operation == "INCREMENT":
# Build increment expression
update_expr_parts = []
expr_attr_values = {}

for field, value in data.items():
update_expr_parts.append(f"{field} = if_not_exists({field}, :zero) + :{field}")
expr_attr_values[f":{field}"] = {'N': str(value)}

expr_attr_values[":zero"] = {'N': '0'}

# Convert key to DynamoDB format
dynamodb_key = {}
for k, v in key.items():
if isinstance(v, str):
dynamodb_key[k] = {'S': v}
elif isinstance(v, (int, float)):
dynamodb_key[k] = {'N': str(v)}

ddb.update_item(
TableName=table,
Key=dynamodb_key,
UpdateExpression=f"SET {', '.join(update_expr_parts)}",
ExpressionAttributeValues=expr_attr_values
)
Comment thread
coderabbitai[bot] marked this conversation as resolved.

else:
# Unknown operation type - fail explicitly
total_failed += 1
error_msg = f"Unknown WAL operation type '{operation}' at sequence {entry_sequence}"
warning(error_msg)
errors.append(error_msg)
# STOP processing to prevent replaying later entries (especially INCREMENTs)
break

total_synced += 1

# Update checkpoint after each successful operation
# This is safe because we STOP on first failure (break above)
if entry_sequence >= 0:
wal_manager.set_last_applied_sequence(entry_sequence)

except Exception as e:
total_failed += 1
error_msg = f"Failed to sync WAL entry at sequence {entry_sequence} to {table}: {e}"
error(error_msg)
errors.append(error_msg)
# STOP processing to prevent replaying later entries (especially INCREMENTs)
break

# Mark success if all synced
if total_failed == 0:
# CRITICAL FIX: Clear WAL entries ONLY up to the last successfully applied sequence
# This prevents race condition where new writes after our snapshot are lost

# Get the highest sequence we successfully applied
last_synced_sequence = wal_manager.get_last_applied_sequence()

# Clear WAL entries up to last_synced_sequence, keeping any concurrent writes
# that occurred after our snapshot (sequence > last_synced_sequence)
wal_manager.clear_up_to(last_synced_sequence)

# Note: We intentionally DON'T call cache_manager.mark_synced() here because:
# - Cache entries don't track their WAL sequence number
# - We can't safely determine which cache entries map to synced WAL entries
# - Dirty flags in cache are eventually consistent (background task marks them synced)
# - The WAL checkpoint is our source of truth for what's been persisted

info(f"✅ Cache dump complete: {total_synced} WAL operations synced to DynamoDB (checkpoint: {last_synced_sequence})")

return {
"success": True,
"entries": total_written,
"results": results
"entries": total_synced,
"failed": 0,
"checkpoint": last_synced_sequence
}
else:
total_failed = sum(r.get('failed', 0) for r in results.values())
error(f"⚠️ Cache dump partially failed: {total_failed} entries failed")
warning(f"⚠️ Cache dump partially failed: {total_failed}/{len(wal_entries)} operations failed")

return {
"success": False,
"entries": len(dirty_entries),
"entries": len(wal_entries),
"synced": total_synced,
"failed": total_failed,
"results": results
"errors": errors
}

except Exception as e:
Expand Down
Loading