Quorum-Based Replication

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
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150

interface Hint {
  id: string;
  targetNodeId: string;       // Where data should eventually go
  key: string;                // The data key
  value: any;                 // The actual data
  writeTimestamp: number;     // Original write time
  hintCreatedAt: number;      // When hint was created
  ttlMs: number;              // How long to keep the hint
  attempts: number;           // Delivery attempts so far
  lastAttempt?: number;       // Timestamp of last delivery attempt
}
 
interface HintStore {
  addHint(hint: Hint): Promise<void>;
  getHints(targetNodeId: string): Promise<Hint[]>;
  getAllHints(): Promise<Hint[]>;
  deleteHint(hintId: string): Promise<void>;
  deleteExpiredHints(): Promise<number>;
}
 
class HintedHandoffManager {
  private hintStore: HintStore;
  private nodeMonitor: NodeMonitor;
  private maxAttempts: number = 10;
  private retryDelayMs: number = 5000;
  private deliveryBatchSize: number = 100;
  
  constructor(hintStore: HintStore, nodeMonitor: NodeMonitor) {
    this.hintStore = hintStore;
    this.nodeMonitor = nodeMonitor;
  }
  
  // Stage 1 & 2: Generate and store hint
  async createHint(
    targetNodeId: string,
    key: string,
    value: any,
    writeTimestamp: number,
    ttlMs: number = 3 * 60 * 60 * 1000  // 3 hours default
  ): Promise<Hint> {
    const hint: Hint = {
      id: `hint-${Date.now()}-${Math.random().toString(36).slice(2)}`,
      targetNodeId,
      key,
      value,
      writeTimestamp,
      hintCreatedAt: Date.now(),
      ttlMs,
      attempts: 0,
    };
    
    await this.hintStore.addHint(hint);
    console.log(`Created hint ${hint.id} for ${targetNodeId}, key=${key}`);
    
    return hint;
  }
  
  // Stage 3: Background delivery process
  async runDeliveryLoop(): Promise<void> {
    while (true) {
      try {
        await this.deliverPendingHints();
        await this.hintStore.deleteExpiredHints();
      } catch (error) {
        console.error('Hint delivery error:', error);
      }
      
      await this.sleep(this.retryDelayMs);
    }
  }
  
  private async deliverPendingHints(): Promise<void> {
    const allHints = await this.hintStore.getAllHints();
    
    // Group hints by target node
    const hintsByTarget = new Map<string, Hint[]>();
    for (const hint of allHints) {
      if (!hintsByTarget.has(hint.targetNodeId)) {
        hintsByTarget.set(hint.targetNodeId, []);
      }
      hintsByTarget.get(hint.targetNodeId)!.push(hint);
    }
    
    // Attempt delivery to each recovered node
    for (const [targetNodeId, hints] of hintsByTarget) {
      const isAvailable = await this.nodeMonitor.isNodeAvailable(targetNodeId);
      
      if (!isAvailable) {
        console.log(`Target ${targetNodeId} still unavailable, skipping ${hints.length} hints`);
        continue;
      }
      
      // Deliver in batches
      for (let i = 0; i < hints.length; i += this.deliveryBatchSize) {
        const batch = hints.slice(i, i + this.deliveryBatchSize);
        await this.deliverBatch(targetNodeId, batch);
      }
    }
  }
  
  private async deliverBatch(targetNodeId: string, hints: Hint[]): Promise<void> {
    console.log(`Delivering ${hints.length} hints to ${targetNodeId}`);
    
    for (const hint of hints) {
      // Check TTL
      if (Date.now() - hint.hintCreatedAt > hint.ttlMs) {
        console.warn(`Hint ${hint.id} expired, discarding`);
        await this.hintStore.deleteHint(hint.id);
        continue;
      }
      
      // Check max attempts
      if (hint.attempts >= this.maxAttempts) {
        console.error(`Hint ${hint.id} exceeded max attempts, discarding`);
        await this.hintStore.deleteHint(hint.id);
        continue;
      }
      
      try {
        // Attempt delivery
        await this.sendToNode(targetNodeId, hint.key, hint.value, hint.writeTimestamp);
        
        // Success! Stage 4: Cleanup
        await this.hintStore.deleteHint(hint.id);
        console.log(`Successfully delivered hint ${hint.id}`);
        
      } catch (error) {
        // Delivery failed, increment attempts
        hint.attempts++;
        hint.lastAttempt = Date.now();
        console.warn(`Failed to deliver hint ${hint.id}, attempt ${hint.attempts}`);
        
        // If node became unavailable during delivery, stop this batch
        if (!await this.nodeMonitor.isNodeAvailable(targetNodeId)) {
          console.log(`Target ${targetNodeId} became unavailable, stopping batch`);
          break;
        }
      }
    }
  }
  
  private async sendToNode(nodeId: string, key: string, value: any, ts: number): Promise<void> {
    // Actual network call to write data to target node
    // Implementation depends on the specific system
  }
  
  private sleep(ms: number): Promise<void> {
    return new Promise(resolve => setTimeout(resolve, ms));
  }
}

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
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181

import fs from 'fs/promises';
import path from 'path';
 
interface SerializedHint {
  id: string;
  targetNodeId: string;
  key: string;
  value: string;  // JSON-serialized
  writeTimestamp: number;
  hintCreatedAt: number;
  ttlMs: number;
  attempts: number;
  lastAttempt?: number;
}
 
class FileBasedHintStore implements HintStore {
  private hintsDirectory: string;
  private memoryIndex: Map<string, SerializedHint> = new Map();
  private dirty: boolean = false;
  
  constructor(hintsDir: string) {
    this.hintsDirectory = hintsDir;
  }
  
  async initialize(): Promise<void> {
    await fs.mkdir(this.hintsDirectory, { recursive: true });
    await this.loadIndex();
    
    // Periodic flush to disk
    setInterval(() => this.flushIfDirty(), 1000);
  }
  
  private async loadIndex(): Promise<void> {
    try {
      const indexPath = path.join(this.hintsDirectory, 'index.json');
      const data = await fs.readFile(indexPath, 'utf-8');
      const hints: SerializedHint[] = JSON.parse(data);
      
      for (const hint of hints) {
        this.memoryIndex.set(hint.id, hint);
      }
      
      console.log(`Loaded ${this.memoryIndex.size} hints from disk`);
    } catch (error: any) {
      if (error.code !== 'ENOENT') throw error;
      console.log('No existing hint index, starting fresh');
    }
  }
  
  private async flushIfDirty(): Promise<void> {
    if (!this.dirty) return;
    
    const indexPath = path.join(this.hintsDirectory, 'index.json');
    const hints = Array.from(this.memoryIndex.values());
    
    // Atomic write: write to temp file then rename
    const tempPath = indexPath + '.tmp';
    await fs.writeFile(tempPath, JSON.stringify(hints, null, 2));
    await fs.rename(tempPath, indexPath);
    
    this.dirty = false;
    console.log(`Flushed ${hints.length} hints to disk`);
  }
  
  async addHint(hint: Hint): Promise<void> {
    const serialized: SerializedHint = {
      ...hint,
      value: JSON.stringify(hint.value),
    };
    
    this.memoryIndex.set(hint.id, serialized);
    this.dirty = true;
    
    // For large values, write to separate files
    if (serialized.value.length > 10000) {
      const valuePath = path.join(this.hintsDirectory, `${hint.id}.value`);
      await fs.writeFile(valuePath, serialized.value);
      serialized.value = `__file__:${hint.id}.value`;
    }
  }
  
  async getHints(targetNodeId: string): Promise<Hint[]> {
    const hints: Hint[] = [];
    
    for (const serialized of this.memoryIndex.values()) {
      if (serialized.targetNodeId === targetNodeId) {
        hints.push(await this.deserialize(serialized));
      }
    }
    
    return hints;
  }
  
  async getAllHints(): Promise<Hint[]> {
    const hints: Hint[] = [];
    
    for (const serialized of this.memoryIndex.values()) {
      hints.push(await this.deserialize(serialized));
    }
    
    return hints;
  }
  
  private async deserialize(serialized: SerializedHint): Promise<Hint> {
    let value = serialized.value;
    
    // Load from file if stored externally
    if (value.startsWith('__file__:')) {
      const fileName = value.slice('__file__:'.length);
      const valuePath = path.join(this.hintsDirectory, fileName);
      value = await fs.readFile(valuePath, 'utf-8');
    }
    
    return {
      ...serialized,
      value: JSON.parse(value),
    };
  }
  
  async deleteHint(hintId: string): Promise<void> {
    const hint = this.memoryIndex.get(hintId);
    if (!hint) return;
    
    // Delete external value file if exists
    if (hint.value.startsWith('__file__:')) {
      const fileName = hint.value.slice('__file__:'.length);
      const valuePath = path.join(this.hintsDirectory, fileName);
      await fs.unlink(valuePath).catch(() => {});
    }
    
    this.memoryIndex.delete(hintId);
    this.dirty = true;
  }
  
  async deleteExpiredHints(): Promise<number> {
    const now = Date.now();
    let deleted = 0;
    
    for (const [id, hint] of this.memoryIndex) {
      if (now - hint.hintCreatedAt > hint.ttlMs) {
        await this.deleteHint(id);
        deleted++;
      }
    }
    
    if (deleted > 0) {
      console.log(`Deleted ${deleted} expired hints`);
    }
    
    return deleted;
  }
  
  // Metrics
  async getStats(): Promise<{
    totalHints: number;
    hintsByTarget: Map<string, number>;
    oldestHintAge: number;
    totalSize: number;
  }> {
    const hintsByTarget = new Map<string, number>();
    let oldestHintAge = 0;
    let totalSize = 0;
    
    for (const hint of this.memoryIndex.values()) {
      const count = hintsByTarget.get(hint.targetNodeId) || 0;
      hintsByTarget.set(hint.targetNodeId, count + 1);
      
      const age = Date.now() - hint.hintCreatedAt;
      if (age > oldestHintAge) oldestHintAge = age;
      
      totalSize += hint.value.length;
    }
    
    return {
      totalHints: this.memoryIndex.size,
      hintsByTarget,
      oldestHintAge,
      totalSize,
    };
  }
}

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
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147

interface DeliveryConfig {
  maxBytesPerSecond: number;      // Bandwidth limit
  maxConcurrentDeliveries: number; // Parallel delivery limit
  batchSize: number;               // Hints per batch
  retryBackoffMs: number[];        // Exponential backoff delays
}
 
class ThrottledHintDelivery {
  private config: DeliveryConfig;
  private activeDeliveries: Set<string> = new Set();
  private bytesDeliveredThisSecond: number = 0;
  private lastSecondReset: number = Date.now();
  
  constructor(config: DeliveryConfig) {
    this.config = config;
    
    // Reset bandwidth counter each second
    setInterval(() => {
      this.bytesDeliveredThisSecond = 0;
      this.lastSecondReset = Date.now();
    }, 1000);
  }
  
  async deliverToNode(
    targetNodeId: string,
    hints: Hint[]
  ): Promise<DeliveryResult> {
    // Check concurrent delivery limit
    if (this.activeDeliveries.size >= this.config.maxConcurrentDeliveries) {
      return { delivered: 0, skipped: hints.length, reason: 'max_concurrent' };
    }
    
    this.activeDeliveries.add(targetNodeId);
    
    try {
      let delivered = 0;
      let failed = 0;
      
      // Process in batches
      for (let i = 0; i < hints.length; i += this.config.batchSize) {
        const batch = hints.slice(i, i + this.config.batchSize);
        const batchResult = await this.deliverBatchWithThrottle(targetNodeId, batch);
        
        delivered += batchResult.delivered;
        failed += batchResult.failed;
        
        // If batch failed completely, stop trying
        if (batchResult.failed === batch.length) {
          return {
            delivered,
            skipped: hints.length - i - batch.length,
            failed,
            reason: 'batch_failed',
          };
        }
      }
      
      return { delivered, failed };
      
    } finally {
      this.activeDeliveries.delete(targetNodeId);
    }
  }
  
  private async deliverBatchWithThrottle(
    targetNodeId: string,
    hints: Hint[]
  ): Promise<{ delivered: number; failed: number }> {
    let delivered = 0;
    let failed = 0;
    
    for (const hint of hints) {
      // Calculate hint size for bandwidth throttling
      const hintSize = JSON.stringify(hint.value).length;
      
      // Wait if we'd exceed bandwidth limit
      while (this.bytesDeliveredThisSecond + hintSize > this.config.maxBytesPerSecond) {
        const msUntilReset = 1000 - (Date.now() - this.lastSecondReset);
        if (msUntilReset > 0) {
          await this.sleep(msUntilReset);
        }
        // Counter will be reset by the interval
      }
      
      // Attempt delivery with retry
      const success = await this.deliverWithRetry(targetNodeId, hint);
      
      if (success) {
        delivered++;
        this.bytesDeliveredThisSecond += hintSize;
      } else {
        failed++;
      }
    }
    
    return { delivered, failed };
  }
  
  private async deliverWithRetry(targetNodeId: string, hint: Hint): Promise<boolean> {
    const backoffs = this.config.retryBackoffMs;
    
    for (let attempt = 0; attempt <= backoffs.length; attempt++) {
      try {
        await this.sendHint(targetNodeId, hint);
        return true;
      } catch (error) {
        if (attempt < backoffs.length) {
          await this.sleep(backoffs[attempt]);
        }
      }
    }
    
    return false;
  }
  
  private async sendHint(targetNodeId: string, hint: Hint): Promise<void> {
    // Actual RPC to target node
    // Implementation depends on transport (gRPC, HTTP, custom protocol)
    console.log(`Sending hint ${hint.id} to ${targetNodeId}`);
  }
  
  private sleep(ms: number): Promise<void> {
    return new Promise(resolve => setTimeout(resolve, ms));
  }
}
 
interface DeliveryResult {
  delivered: number;
  skipped?: number;
  failed?: number;
  reason?: string;
}
 
// Configuration for different scenarios
const conservativeConfig: DeliveryConfig = {
  maxBytesPerSecond: 1024 * 1024,    // 1 MB/s
  maxConcurrentDeliveries: 2,
  batchSize: 50,
  retryBackoffMs: [100, 500, 1000, 5000],
};
 
const aggressiveConfig: DeliveryConfig = {
  maxBytesPerSecond: 10 * 1024 * 1024,  // 10 MB/s
  maxConcurrentDeliveries: 10,
  batchSize: 500,
  retryBackoffMs: [50, 100, 500],
};

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
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126

class RobustHintedHandoffManager extends HintedHandoffManager {
  private clusterMembership: ClusterMembershipService;
  private antiEntropy: AntiEntropyService;
  
  // Handle node replacement
  async handleNodeReplacement(oldNodeId: string, newNodeId: string): Promise<void> {
    console.log(`Node replacement: ${oldNodeId} -> ${newNodeId}`);
    
    // Get all hints destined for the old node
    const orphanedHints = await this.hintStore.getHints(oldNodeId);
    
    if (orphanedHints.length === 0) {
      console.log('No orphaned hints to migrate');
      return;
    }
    
    console.log(`Migrating ${orphanedHints.length} hints to ${newNodeId}`);
    
    // Update hint targets
    for (const hint of orphanedHints) {
      // Delete old hint
      await this.hintStore.deleteHint(hint.id);
      
      // Create new hint for replacement node
      await this.createHint(
        newNodeId,
        hint.key,
        hint.value,
        hint.writeTimestamp,
        hint.ttlMs - (Date.now() - hint.hintCreatedAt)  // Preserve remaining TTL
      );
    }
  }
  
  // Handle hint store pressure
  async handleDiskPressure(): Promise<void> {
    const stats = await (this.hintStore as any).getStats();
    const thresholds = {
      warningPercent: 70,
      criticalPercent: 90,
    };
    
    const usagePercent = await this.getDiskUsagePercent();
    
    if (usagePercent > thresholds.criticalPercent) {
      console.error(`CRITICAL: Hint store at ${usagePercent}% capacity`);
      
      // Emergency: delete oldest hints to free space
      const oldestHints = await this.getOldestHints(100);
      for (const hint of oldestHints) {
        await this.hintStore.deleteHint(hint.id);
      }
      
      console.warn('Deleted oldest hints to recover space - data may be lost');
      
    } else if (usagePercent > thresholds.warningPercent) {
      console.warn(`WARNING: Hint store at ${usagePercent}% capacity`);
      
      // Accelerate delivery to healthy nodes
      await this.prioritizeDelivery();
    }
  }
  
  // Handle prolonged outages with anti-entropy
  async handleProlongedOutage(nodeId: string, outageMs: number): Promise<void> {
    const hintTtlMs = 3 * 60 * 60 * 1000;  // 3 hours
    
    if (outageMs > hintTtlMs * 0.8) {
      console.warn(`Node ${nodeId} approaching hint TTL, scheduling anti-entropy`);
      
      // Schedule full repair when node recovers
      this.clusterMembership.onNodeRecovery(nodeId, async () => {
        console.log(`Running anti-entropy repair for ${nodeId}`);
        
        // Anti-entropy will catch any data missed due to expired hints
        await this.antiEntropy.repairNode(nodeId);
      });
    }
  }
  
  // Handle duplicate hints (optimization)
  async compactHints(): Promise<number> {
    const allHints = await this.hintStore.getAllHints();
    
    // Group by target and key
    const latestByKey = new Map<string, Hint>();
    const toDelete: string[] = [];
    
    for (const hint of allHints) {
      const compoundKey = `${hint.targetNodeId}:${hint.key}`;
      const existing = latestByKey.get(compoundKey);
      
      if (!existing || hint.writeTimestamp > existing.writeTimestamp) {
        if (existing) {
          toDelete.push(existing.id);
        }
        latestByKey.set(compoundKey, hint);
      } else {
        toDelete.push(hint.id);
      }
    }
    
    for (const hintId of toDelete) {
      await this.hintStore.deleteHint(hintId);
    }
    
    console.log(`Compacted ${toDelete.length} duplicate hints`);
    return toDelete.length;
  }
  
  private async getDiskUsagePercent(): Promise<number> {
    // Implementation depends on OS and storage
    return 0;
  }
  
  private async getOldestHints(count: number): Promise<Hint[]> {
    const allHints = await this.hintStore.getAllHints();
    return allHints
      .sort((a, b) => a.hintCreatedAt - b.hintCreatedAt)
      .slice(0, count);
  }
  
  private async prioritizeDelivery(): Promise<void> {
    // Increase delivery threads/bandwidth temporarily
  }
}

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
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

import crypto from 'crypto';
 
interface MerkleNode {
  hash: string;
  left?: MerkleNode;
  right?: MerkleNode;
  keyRange?: { start: string; end: string };  // For leaf nodes
}
 
class MerkleTree {
  private root: MerkleNode | null = null;
  private leafCount: number;
  
  constructor(leafCount: number = 1024) {
    this.leafCount = leafCount;
  }
  
  // Build tree from key-value data
  async buildFromData(data: Map<string, any>): Promise<void> {
    // Sort keys and divide into buckets
    const sortedKeys = [...data.keys()].sort();
    const buckets = this.divideToBuckets(sortedKeys, this.leafCount);
    
    // Create leaf nodes
    const leaves: MerkleNode[] = [];
    for (const bucket of buckets) {
      const bucketData = bucket.keys.map(k => `${k}:${JSON.stringify(data.get(k))}`).join('|');
      leaves.push({
        hash: this.hash(bucketData),
        keyRange: { start: bucket.keys[0] || '', end: bucket.keys[bucket.keys.length - 1] || '' },
      });
    }
    
    // Build tree from leaves up
    this.root = this.buildTreeFromLeaves(leaves);
  }
  
  private divideToBuckets(keys: string[], bucketCount: number): { keys: string[] }[] {
    const buckets: { keys: string[] }[] = [];
    const keysPerBucket = Math.ceil(keys.length / bucketCount);
    
    for (let i = 0; i < bucketCount; i++) {
      buckets.push({
        keys: keys.slice(i * keysPerBucket, (i + 1) * keysPerBucket),
      });
    }
    
    return buckets;
  }
  
  private buildTreeFromLeaves(leaves: MerkleNode[]): MerkleNode {
    if (leaves.length === 1) return leaves[0];
    
    const parents: MerkleNode[] = [];
    
    for (let i = 0; i < leaves.length; i += 2) {
      const left = leaves[i];
      const right = leaves[i + 1] || left;  // Duplicate if odd number
      
      parents.push({
        hash: this.hash(left.hash + right.hash),
        left,
        right: leaves[i + 1] ? right : undefined,
      });
    }
    
    return this.buildTreeFromLeaves(parents);
  }
  
  private hash(data: string): string {
    return crypto.createHash('sha256').update(data).digest('hex').slice(0, 16);
  }
  
  getRootHash(): string {
    return this.root?.hash || '';
  }
  
  // Find differences between local and remote trees
  async findDifferences(
    localTree: MerkleTree,
    remoteTree: MerkleTree,
    exchangeHash: (path: string) => Promise<string>
  ): Promise<{ start: string; end: string }[]> {
    const differences: { start: string; end: string }[] = [];
    
    async function compare(
      local: MerkleNode | undefined,
      path: string
    ): Promise<void> {
      if (!local) return;
      
      const remoteHash = await exchangeHash(path);
      
      if (local.hash === remoteHash) {
        // Subtrees are identical
        return;
      }
      
      if (local.keyRange) {
        // This is a leaf node - found a difference
        differences.push(local.keyRange);
        return;
      }
      
      // Recursively check children
      await compare(local.left, path + '0');
      await compare(local.right, path + '1');
    }
    
    await compare(localTree.root, '');
    
    return differences;
  }
}
 
// Usage in anti-entropy repair
async function repairWithMerkleTree(
  localNode: Node,
  remoteNode: Node
): Promise<void> {
  // Build local tree
  const localData = await localNode.getAllData();
  const localTree = new MerkleTree();
  await localTree.buildFromData(localData);
  
  // Exchange hashes with remote
  const remoteTree = new MerkleTree();  // Would be on remote node
  
  const differences = await localTree.findDifferences(
    localTree,
    remoteTree,
    async (path) => {
      // RPC to get hash at path from remote
      return remoteNode.getMerkleHash(path);
    }
  );
  
  console.log(`Found ${differences.length} differing ranges`);
  
  // Synchronize only the differing ranges
  for (const range of differences) {
    const remoteData = await remoteNode.getDataInRange(range.start, range.end);
    await localNode.reconcile(remoteData);
  }
}

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
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137

interface HintMetrics {
  timestamp: number;
  totalHints: number;
  hintsByTarget: Map<string, number>;
  creationRate: number;  // hints/sec
  deliveryRate: number;  // hints/sec
  oldestHintAge: number; // ms
  expirationCount: number;
  diskUsagePercent: number;
}
 
class HintMonitoringService {
  private metrics: HintMetrics[] = [];
  private alertThresholds = {
    totalHints: 500000,
    hintsPerTarget: 100000,
    creationVsDeliveryRatio: 1.5,
    oldestHintAgePct: 0.5,  // % of TTL
    expirations: 1,
    diskUsage: 70,
  };
  private hintTtlMs: number;
  
  constructor(hintTtlMs: number = 3 * 60 * 60 * 1000) {
    this.hintTtlMs = hintTtlMs;
  }
  
  async collectMetrics(hintStore: HintStore): Promise<HintMetrics> {
    const stats = await (hintStore as any).getStats();
    
    const metrics: HintMetrics = {
      timestamp: Date.now(),
      totalHints: stats.totalHints,
      hintsByTarget: stats.hintsByTarget,
      creationRate: this.calculateRate('creation'),
      deliveryRate: this.calculateRate('delivery'),
      oldestHintAge: stats.oldestHintAge,
      expirationCount: await this.getExpirationCount(hintStore),
      diskUsagePercent: await this.getDiskUsage(),
    };
    
    this.metrics.push(metrics);
    
    // Check alerts
    this.evaluateAlerts(metrics);
    
    return metrics;
  }
  
  private calculateRate(type: 'creation' | 'delivery'): number {
    // Calculate from recent metrics
    // Implementation depends on metric collection system
    return 0;
  }
  
  private async getExpirationCount(hintStore: HintStore): Promise<number> {
    // Would be tracked by the store
    return 0;
  }
  
  private async getDiskUsage(): Promise<number> {
    // OS-specific implementation
    return 0;
  }
  
  private evaluateAlerts(metrics: HintMetrics): void {
    const alerts: string[] = [];
    
    // Total hints
    if (metrics.totalHints > this.alertThresholds.totalHints) {
      alerts.push(`CRITICAL: Total hints (${metrics.totalHints}) exceeds threshold`);
    }
    
    // Per-target hints
    for (const [target, count] of metrics.hintsByTarget) {
      if (count > this.alertThresholds.hintsPerTarget) {
        alerts.push(`WARNING: Hints for ${target} (${count}) exceeds threshold`);
      }
    }
    
    // Creation vs delivery rate
    if (metrics.creationRate > metrics.deliveryRate * this.alertThresholds.creationVsDeliveryRatio) {
      alerts.push(
        `WARNING: Hint creation (${metrics.creationRate}/s) outpacing delivery (${metrics.deliveryRate}/s)`
      );
    }
    
    // Oldest hint age
    const agePct = metrics.oldestHintAge / this.hintTtlMs;
    if (agePct > this.alertThresholds.oldestHintAgePct) {
      alerts.push(
        `WARNING: Oldest hint at ${(agePct * 100).toFixed(1)}% of TTL. Node may not recover in time.`
      );
    }
    
    // Expirations
    if (metrics.expirationCount > this.alertThresholds.expirations) {
      alerts.push(
        `CRITICAL: ${metrics.expirationCount} hints expired. DATA LOSS MAY HAVE OCCURRED.`
      );
    }
    
    // Disk usage
    if (metrics.diskUsagePercent > this.alertThresholds.diskUsage) {
      alerts.push(
        `WARNING: Hint disk usage at ${metrics.diskUsagePercent}%. May reject new hints soon.`
      );
    }
    
    if (alerts.length > 0) {
      console.log('=== HINT HANDOFF ALERTS ===');
      alerts.forEach(a => console.log(a));
      
      // Send to alerting system
      this.sendAlerts(alerts);
    }
  }
  
  private sendAlerts(alerts: string[]): void {
    // Integration with PagerDuty, Slack, etc.
  }
  
  // Runbook actions
  async runRunbook(action: 'prioritize_delivery' | 'expand_disk' | 'trigger_repair'): Promise<void> {
    switch (action) {
      case 'prioritize_delivery':
        console.log('ACTION: Increasing hint delivery threads and bandwidth');
        break;
      case 'expand_disk':
        console.log('ACTION: Expanding hint storage capacity');
        break;
      case 'trigger_repair':
        console.log('ACTION: Scheduling anti-entropy repair for affected nodes');
        break;
    }
  }
}

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
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107

interface VersionedValue {
  value: any;
  version: number;
  timestamp: number;
  nodeId: string;
}
 
class ReadRepairCoordinator {
  async readWithRepair(
    key: string,
    replicas: Node[],
    readQuorum: number
  ): Promise<any> {
    // Query all replicas
    const responses = await Promise.allSettled(
      replicas.map(async r => {
        const result = await r.read(key);
        return { ...result, nodeId: r.id };
      })
    );
    
    const successResponses = responses
      .filter((r): r is PromiseFulfilledResult<VersionedValue> => r.status === 'fulfilled')
      .map(r => r.value);
    
    if (successResponses.length < readQuorum) {
      throw new Error(`Read quorum not met: ${successResponses.length}/${readQuorum}`);
    }
    
    // Find the latest version
    const latest = successResponses.reduce((a, b) => 
      a.timestamp > b.timestamp ? a : b
    );
    
    // Identify stale replicas
    const staleReplicas = successResponses
      .filter(r => r.version !== latest.version)
      .map(r => r.nodeId);
    
    // Trigger async read repair for stale replicas
    if (staleReplicas.length > 0) {
      this.triggerReadRepair(key, latest, staleReplicas);
    }
    
    return latest.value;
  }
  
  private async triggerReadRepair(
    key: string,
    latestValue: VersionedValue,
    staleNodeIds: string[]
  ): Promise<void> {
    // Fire-and-forget repair
    setImmediate(async () => {
      console.log(`Read repair: updating ${staleNodeIds.length} stale replicas for ${key}`);
      
      for (const nodeId of staleNodeIds) {
        try {
          await this.writeToNode(nodeId, key, latestValue);
          console.log(`Repaired ${nodeId} for key ${key}`);
        } catch (error) {
          console.warn(`Failed to repair ${nodeId}: ${error}`);
          // Could create a hint if the repair fails
        }
      }
    });
  }
  
  private async writeToNode(nodeId: string, key: string, value: VersionedValue): Promise<void> {
    // RPC to write data
  }
}
 
// Combined healing strategy
class ConsistencyHealingStrategy {
  private hintManager: HintedHandoffManager;
  private readRepair: ReadRepairCoordinator;
  private antiEntropy: AntiEntropyService;
  
  /**
   * Choose the appropriate healing mechanism based on the situation
   */
  async heal(
    key: string,
    context: 'write_failure' | 'read_inconsistency' | 'scheduled_repair'
  ): Promise<void> {
    switch (context) {
      case 'write_failure':
        // Node is down during write - create hint
        console.log('Creating hint for later delivery');
        // Hinted handoff will deliver when node recovers
        break;
        
      case 'read_inconsistency':
        // Found stale data during read - repair immediately
        console.log('Triggering read repair');
        // Read repair updates stale replicas
        break;
        
      case 'scheduled_repair':
        // Proactive consistency check
        console.log('Running anti-entropy repair');
        // Merkle tree comparison finds all inconsistencies
        break;
    }
  }
}