Detection

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

-- Invariant verification: account balance should equal sum of transactions
WITH expected AS (
    SELECT 
        account_id,
        (SELECT initial_deposit FROM account_setup WHERE account_id = t.account_id) +
        SUM(CASE WHEN type = 'CREDIT' THEN amount ELSE -amount END) as expected_balance
    FROM transactions t
    GROUP BY account_id
),
actual AS (
    SELECT account_id, balance as actual_balance
    FROM accounts
)
SELECT 
    e.account_id,
    e.expected_balance,
    a.actual_balance,
    a.actual_balance - e.expected_balance as discrepancy
FROM expected e
JOIN actual a ON e.account_id = a.account_id
WHERE a.actual_balance != e.expected_balance;
 
-- If this query returns rows, we have a discrepancy
-- Lost updates are a primary suspect
 
-- Example output:
-- account_id | expected_balance | actual_balance | discrepancy
-- ACC-7821   | 0.00             | 100.00         | +100.00
-- ACC-9432   | 1500.00          | 1520.00        | +20.00

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

-- Request logs table: records all incoming increment requests
CREATE TABLE request_log (
    request_id UUID PRIMARY KEY,
    timestamp TIMESTAMP NOT NULL,
    entity_type VARCHAR(50) NOT NULL,
    entity_id VARCHAR(100) NOT NULL,
    operation VARCHAR(50) NOT NULL,
    amount INT NOT NULL
);
 
-- Example: page view increment requests
-- Every time a user views a page, we log the request
 
-- Modification logs table: records successful database changes
CREATE TABLE modification_log (
    mod_id SERIAL PRIMARY KEY,
    request_id UUID,  -- Links to the triggering request
    timestamp TIMESTAMP NOT NULL,
    table_name VARCHAR(100) NOT NULL,
    record_id VARCHAR(100) NOT NULL,
    old_value TEXT,
    new_value TEXT
);
 
-- Detection query: find requests without corresponding modifications
SELECT 
    r.request_id,
    r.timestamp as request_time,
    r.entity_id,
    r.operation,
    r.amount
FROM request_log r
LEFT JOIN modification_log m ON r.request_id = m.request_id
WHERE r.entity_type = 'page_view'
  AND r.timestamp > NOW() - INTERVAL '24 hours'
  AND m.request_id IS NULL;
 
-- Results show page view requests that did not produce modifications
-- These are likely lost updates

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

-- Compare request log count against actual counter value
WITH request_counts AS (
    SELECT 
        entity_id as page_id,
        COUNT(*) as request_count
    FROM request_log
    WHERE entity_type = 'page_view'
      AND timestamp BETWEEN '2024-01-01' AND '2024-01-31'
    GROUP BY entity_id
),
actual_counts AS (
    SELECT 
        page_id,
        view_count - lag_view_count as increment_count
    FROM (
        SELECT 
            page_id,
            view_count,
            LAG(view_count) OVER (PARTITION BY page_id ORDER BY snapshot_date) as lag_view_count
        FROM page_count_snapshots
        WHERE snapshot_date IN ('2024-01-01', '2024-02-01')
    ) sub
    WHERE lag_view_count IS NOT NULL
)
SELECT 
    r.page_id,
    r.request_count as expected_increments,
    a.increment_count as actual_increments,
    r.request_count - a.increment_count as lost_updates,
    ROUND(100.0 * (r.request_count - a.increment_count) / r.request_count, 2) as loss_percentage
FROM request_counts r
JOIN actual_counts a ON r.page_id = a.page_id
WHERE r.request_count > a.increment_count
ORDER BY lost_updates DESC;
 
-- Sample output:
-- page_id   | expected | actual | lost_updates | loss_pct
-- P-12345   | 50000    | 47500  | 2500         | 5.00%
-- P-67890   | 30000    | 28800  | 1200         | 4.00%

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

-- Drift analysis: compare expected growth rate vs actual
-- If a counter should grow roughly linearly but shows consistent under-performance...
 
WITH daily_metrics AS (
    SELECT 
        metric_date,
        daily_requests,  -- From web server logs
        daily_increments, -- From database counter delta
        daily_requests - daily_increments as daily_loss
    FROM metric_summary
    WHERE metric_name = 'page_views'
    ORDER BY metric_date
),
cumulative AS (
    SELECT 
        metric_date,
        SUM(daily_requests) OVER (ORDER BY metric_date) as cum_requests,
        SUM(daily_increments) OVER (ORDER BY metric_date) as cum_increments,
        SUM(daily_loss) OVER (ORDER BY metric_date) as cum_loss
    FROM daily_metrics
)
SELECT 
    metric_date,
    cum_requests,
    cum_increments,
    cum_loss,
    ROUND(100.0 * cum_loss / cum_requests, 2) as cumulative_loss_pct
FROM cumulative
ORDER BY metric_date;
 
-- If cumulative_loss_pct increases steadily over time,
-- this indicates ongoing lost updates, not one-time errors

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

import asyncio
import aiohttp
import random
 
async def test_lost_updates():
    """
    Test for lost updates by sending concurrent increment requests
    and verifying the final count matches the request count.
    """
    COUNTER_ID = "test-counter-001"
    NUM_REQUESTS = 1000
    CONCURRENT_REQUESTS = 50  # High concurrency to trigger race conditions
    
    # Reset counter to 0
    async with aiohttp.ClientSession() as session:
        await session.post(
            f"http://localhost:8080/counters/{COUNTER_ID}/reset"
        )
    
    # Send concurrent increment requests
    async def increment():
        async with aiohttp.ClientSession() as session:
            response = await session.post(
                f"http://localhost:8080/counters/{COUNTER_ID}/increment"
            )
            return response.status == 200
    
    # Create batches of concurrent requests
    successful_requests = 0
    for batch in range(NUM_REQUESTS // CONCURRENT_REQUESTS):
        tasks = [increment() for _ in range(CONCURRENT_REQUESTS)]
        results = await asyncio.gather(*tasks)
        successful_requests += sum(results)
    
    # Verify final counter value
    async with aiohttp.ClientSession() as session:
        response = await session.get(
            f"http://localhost:8080/counters/{COUNTER_ID}"
        )
        counter_data = await response.json()
        final_count = counter_data['value']
    
    # Analyze results
    print(f"Total successful requests: {successful_requests}")
    print(f"Final counter value: {final_count}")
    print(f"Lost updates: {successful_requests - final_count}")
    
    if final_count < successful_requests:
        print("❌ LOST UPDATES DETECTED!")
        print(f"   Loss rate: {100 * (successful_requests - final_count) / successful_requests:.2f}%")
        return False
    else:
        print("✓ No lost updates detected")
        return True
 
# Run test
asyncio.run(test_lost_updates())

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81

import threading
import time
 
def test_deterministic_lost_update():
    """
    Uses threading barriers to force the exact R→R→W→W sequence
    that causes lost updates.
    """
    # Setup: initial balance = 1000
    db.execute("UPDATE accounts SET balance = 1000 WHERE id = 'TEST'")
    
    # Barriers to synchronize threads
    both_read = threading.Barrier(2)
    t1_written = threading.Event()
    
    results = {'t1': None, 't2': None}
    
    def transaction_1():
        conn = db.connect()
        cursor = conn.cursor()
        
        cursor.execute("BEGIN")
        cursor.execute("SELECT balance FROM accounts WHERE id = 'TEST'")
        balance = cursor.fetchone()[0]  # Reads 1000
        
        # Wait for T2 to also read
        both_read.wait()
        
        # T1 writes first
        new_balance = balance - 100  # 900
        cursor.execute(f"UPDATE accounts SET balance = {new_balance} WHERE id = 'TEST'")
        cursor.execute("COMMIT")
        
        t1_written.set()  # Signal T2 that T1 has written
        results['t1'] = new_balance
    
    def transaction_2():
        conn = db.connect()
        cursor = conn.cursor()
        
        cursor.execute("BEGIN")
        cursor.execute("SELECT balance FROM accounts WHERE id = 'TEST'")
        balance = cursor.fetchone()[0]  # Also reads 1000 (stale!)
        
        # Wait for T1 to also read
        both_read.wait()
        
        # Wait for T1 to write first
        t1_written.wait()
        
        # T2 writes after T1 - using stale value!
        new_balance = balance - 200  # 800, should be 700!
        cursor.execute(f"UPDATE accounts SET balance = {new_balance} WHERE id = 'TEST'")
        cursor.execute("COMMIT")
        
        results['t2'] = new_balance
    
    t1 = threading.Thread(target=transaction_1)
    t2 = threading.Thread(target=transaction_2)
    
    t1.start()
    t2.start()
    t1.join()
    t2.join()
    
    # Verify final state
    final_balance = db.query("SELECT balance FROM accounts WHERE id = 'TEST'")[0][0]
    expected_balance = 700  # 1000 - 100 - 200
    
    print(f"T1 wrote: {results['t1']}")
    print(f"T2 wrote: {results['t2']}")
    print(f"Final balance: {final_balance}")
    print(f"Expected balance: {expected_balance}")
    
    if final_balance != expected_balance:
        print(f"❌ LOST UPDATE: T1's update was overwritten!")
        print(f"   Lost amount: ${expected_balance - final_balance}")
    else:
        print("✓ Updates correctly serialized")
 
test_deterministic_lost_update()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

# Example Prometheus/Grafana monitoring configuration
 
# Metric: Invariant violations detected
- name: invariant_violations_total
  type: counter
  help: "Total invariant violations detected by verification jobs"
  labels:
    - invariant_type
    - severity
  alerts:
    - name: InvariantViolationDetected
      condition: rate(invariant_violations_total[5m]) > 0
      severity: critical
      message: "Invariant violation detected: {{ $labels.invariant_type }}"
 
# Metric: Request-to-Write discrepancy
- name: request_write_discrepancy_ratio
  type: gauge
  help: "Ratio of logged requests to recorded writes (should be ~1.0)"
  labels:
    - operation_type
  alerts:
    - name: PossibleLostUpdates
      condition: request_write_discrepancy_ratio < 0.99
      for: 10m
      severity: warning
      message: "Possible lost updates: {{ $value | humanizePercentage }} write rate"
 
# Metric: Optimistic lock retries
- name: optimistic_lock_retries_total
  type: counter
  help: "Total optimistic lock conflicts requiring retry"
  labels:
    - table_name
  alerts:
    - name: HighOptimisticLockContention
      condition: rate(optimistic_lock_retries_total[5m]) > 100
      severity: warning
      message: "High contention on {{ $labels.table_name }}: consider pessimistic locking"
 
# Dashboard panel: Lost Update Risk Score
- name: lost_update_risk_score
  type: gauge
  calculation: |
    (1 - request_write_discrepancy_ratio) * 0.4 +
    (rate(optimistic_lock_retries_total[5m]) / 1000) * 0.3 +
    (rate(invariant_violations_total[5m])) * 0.3
  thresholds:
    - value: 0.0
      color: green
      label: "Low Risk"
    - value: 0.1
      color: yellow
      label: "Elevated Risk"
    - value: 0.3
      color: red
      label: "High Risk - Investigate"

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

-- Forensic query: Find overlapping transactions on a specific record
-- Investigating account 'ACC-7821' for potential lost update
 
-- Step 1: Get all transaction log entries for this account in time window
SELECT 
    transaction_id,
    start_time,
    end_time,
    operation,
    old_value,
    new_value
FROM transaction_log
WHERE table_name = 'accounts'
  AND record_id = 'ACC-7821'
  AND start_time BETWEEN '2024-01-15 09:00:00' AND '2024-01-15 09:10:00'
ORDER BY start_time;
 
-- Step 2: Identify overlapping transactions (start_A < end_B AND start_B < end_A)
WITH txns AS (
    SELECT 
        transaction_id as txn,
        start_time,
        end_time
    FROM transaction_log
    WHERE table_name = 'accounts'
      AND record_id = 'ACC-7821'
      AND start_time BETWEEN '2024-01-15 09:00:00' AND '2024-01-15 09:10:00'
)
SELECT 
    a.txn as txn_a,
    b.txn as txn_b,
    a.start_time as a_start,
    a.end_time as a_end,
    b.start_time as b_start,
    b.end_time as b_end
FROM txns a
JOIN txns b ON a.txn < b.txn  -- Avoid duplicate pairs
WHERE a.start_time < b.end_time AND b.start_time < a.end_time;
-- Rows here indicate overlapping transactions - potential lost update
 
-- Step 3: Check if overlapping transactions both read before either wrote
-- (This requires more detailed operation-level logging)
 
-- Step 4: Reconstruct correct value from transaction history
SELECT 
    1000.00 + 
    SUM(CASE WHEN operation = 'CREDIT' THEN amount ELSE -amount END) as correct_balance
FROM transaction_operations
WHERE account_id = 'ACC-7821'
  AND committed = true;

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

# ❌ VULNERABLE: Classic lost update pattern
def vulnerable_increment(item_id):
    # Red flag: SELECT then UPDATE without protection
    item = db.query("SELECT count FROM items WHERE id = ?", item_id)
    new_count = item['count'] + 1
    db.execute("UPDATE items SET count = ? WHERE id = ?", new_count, item_id)
 
# ✓ FIXED: Atomic operation
def atomic_increment(item_id):
    db.execute("UPDATE items SET count = count + 1 WHERE id = ?", item_id)
 
# ❌ VULNERABLE: ORM without optimistic locking
def vulnerable_orm_update(item_id):
    # Red flag: load-modify-save without version checking
    item = Item.get(item_id)
    item.quantity -= 1
    item.save()
 
# ✓ FIXED: ORM with optimistic locking
def safe_orm_update(item_id):
    item = Item.get(item_id)
    item.quantity -= 1
    try:
        item.save()  # Raises StaleObjectError if version changed
    except StaleObjectError:
        # Retry with fresh data
        return safe_orm_update(item_id)
 
# ❌ VULNERABLE: Check-then-act
def vulnerable_booking(room_id, user_id):
    # Red flag: gap between check and action
    room = db.query("SELECT available FROM rooms WHERE id = ?", room_id)
    if room['available']:
        db.execute("UPDATE rooms SET available=false, user=? WHERE id=?", user_id, room_id)
        return True
    return False
 
# ✓ FIXED: Atomic conditional update
def safe_booking(room_id, user_id):
    result = db.execute(
        "UPDATE rooms SET available=false, user=? WHERE id=? AND available=true",
        user_id, room_id
    )
    return result.rowcount == 1  # True if booking succeeded