brainy/docs/brainy-distributed-enhancements.md

Proposed Brainy Enhancements for Distributed Operations

Executive Summary

To fully support the distributed deployment scenario with multiple specialized instances sharing S3 storage, Brainy needs several enhancements focused on coordination, consistency, and operational modes.

Core Enhancement Areas

1. Instance Role Management

Current State: Brainy operates as a standalone instance without awareness of other instances.

Proposed Enhancement:

// New distributed configuration options
export interface DistributedConfig {
  role: 'reader' | 'writer' | 'hybrid';
  instanceId: string;
  coordinationMethod: 'none' | 's3-polling' | 'websocket' | 'pubsub';
  consistencyLevel: 'eventual' | 'strong' | 'bounded';
  conflictResolution: 'last-write-wins' | 'vector-clock' | 'crdt';
}

// Enhanced BrainyData constructor
class BrainyData {
  constructor(config: BrainyConfig & { distributed?: DistributedConfig }) {
    if (config.distributed) {
      this.initializeDistributedMode(config.distributed);
    }
  }
  
  private initializeDistributedMode(config: DistributedConfig) {
    // Set up role-specific behaviors
    switch(config.role) {
      case 'reader':
        this.storage.setReadOnly(true);
        this.enableAggressiveCaching();
        this.subscribeToIndexUpdates();
        break;
      case 'writer':
        this.storage.setWriteOnly(true);
        this.enableWriteBatching();
        this.publishIndexUpdates();
        break;
      case 'hybrid':
        this.enableCoordinatedAccess();
        break;
    }
  }
}

2. S3 Coordination Layer

Current State: Direct S3 operations without coordination.

Proposed Enhancement:

// src/storage/s3Coordinator.ts
export class S3Coordinator {
  private manifestPath = '_brainy/manifest.json';
  private lockPrefix = '_brainy/locks/';
  
  async acquireWriteLock(partition: string): Promise<LockHandle> {
    const lockKey = `${this.lockPrefix}${partition}`;
    const lockId = `${this.instanceId}-${Date.now()}`;
    
    // Use S3's conditional PUT for atomic lock acquisition
    try {
      await this.s3.putObject({
        Bucket: this.bucket,
        Key: lockKey,
        Body: JSON.stringify({
          owner: this.instanceId,
          acquired: Date.now(),
          ttl: 30000  // 30 second TTL
        }),
        Condition: 'ObjectDoesNotExist'
      });
      
      return new LockHandle(lockId, () => this.releaseLock(lockKey));
    } catch (err) {
      if (err.code === 'PreconditionFailed') {
        throw new LockAcquisitionError(`Partition ${partition} is locked`);
      }
      throw err;
    }
  }
  
  async updateManifest(update: ManifestUpdate) {
    // Atomic manifest updates using versioning
    const manifest = await this.getManifest();
    manifest.version++;
    manifest.lastUpdate = Date.now();
    manifest.updates.push(update);
    
    await this.s3.putObject({
      Bucket: this.bucket,
      Key: this.manifestPath,
      Body: JSON.stringify(manifest),
      Metadata: {
        'version': manifest.version.toString()
      }
    });
    
    // Notify other instances
    await this.broadcastUpdate(update);
  }
}

3. Partition Assignment Strategy

Current State: All instances access all partitions.

Proposed Enhancement:

// src/partitioning/distributedPartitioner.ts
export class DistributedPartitioner {
  private assignments: Map<string, Set<string>> = new Map();
  
  async assignPartitions(instances: Instance[]): Promise<PartitionAssignment> {
    const writers = instances.filter(i => i.role === 'writer');
    const partitions = await this.getAllPartitions();
    
    // Use consistent hashing for stable assignments
    const ring = new ConsistentHashRing(writers.map(w => w.id));
    
    const assignment: PartitionAssignment = {};
    for (const partition of partitions) {
      const writer = ring.getNode(partition.id);
      assignment[writer] = assignment[writer] || [];
      assignment[writer].push(partition.id);
    }
    
    // Store assignments in S3 for coordination
    await this.storeAssignments(assignment);
    
    return assignment;
  }
  
  async getMyPartitions(): Promise<string[]> {
    const assignments = await this.loadAssignments();
    return assignments[this.instanceId] || [];
  }
}

4. Write-Ahead Log for Consistency

Current State: Direct writes without transaction log.

Proposed Enhancement:

// src/wal/writeAheadLog.ts
export class WriteAheadLog {
  private logPrefix = '_brainy/wal/';
  
  async logWrite(operation: WriteOperation): Promise<string> {
    const logEntry = {
      id: uuidv4(),
      timestamp: Date.now(),
      instanceId: this.instanceId,
      operation: operation,
      status: 'pending'
    };
    
    // Write to WAL first
    await this.s3.putObject({
      Bucket: this.bucket,
      Key: `${this.logPrefix}${logEntry.id}`,
      Body: JSON.stringify(logEntry)
    });
    
    // Then execute operation
    try {
      await this.executeOperation(operation);
      await this.markComplete(logEntry.id);
    } catch (err) {
      await this.markFailed(logEntry.id, err);
      throw err;
    }
    
    return logEntry.id;
  }
  
  async recoverFromWAL() {
    // On startup, check for incomplete operations
    const pendingOps = await this.getPendingOperations();
    
    for (const op of pendingOps) {
      if (this.canRecover(op)) {
        await this.retryOperation(op);
      }
    }
  }
}

5. Operational Modes

Current State: Single operational mode.

Proposed Enhancement:

// src/modes/operationalModes.ts
export abstract class OperationalMode {
  abstract canRead(): boolean;
  abstract canWrite(): boolean;
  abstract canDelete(): boolean;
  abstract getCacheStrategy(): CacheStrategy;
}

export class ReadOnlyMode extends OperationalMode {
  canRead() { return true; }
  canWrite() { return false; }
  canDelete() { return false; }
  
  getCacheStrategy() {
    return {
      hotCacheRatio: 0.8,  // More memory for cache
      prefetchAggressive: true,
      ttl: Infinity  // Never expire cache in read-only
    };
  }
}

export class WriteOnlyMode extends OperationalMode {
  canRead() { return false; }
  canWrite() { return true; }
  canDelete() { return true; }
  
  getCacheStrategy() {
    return {
      hotCacheRatio: 0.2,  // Minimal cache, focus on write buffer
      writeBuffer: true,
      batchWrites: true
    };
  }
}

export class HybridMode extends OperationalMode {
  constructor(private coordinator: Coordinator) {}
  
  canRead() { return true; }
  canWrite() { return this.coordinator.hasWriteLock(); }
  canDelete() { return this.coordinator.hasWriteLock(); }
  
  getCacheStrategy() {
    return {
      hotCacheRatio: 0.5,
      adaptive: true  // Adjust based on workload
    };
  }
}

6. Health and Coordination Endpoints

Current State: No built-in health/coordination endpoints.

Proposed Enhancement:

// src/api/coordinationAPI.ts
export class CoordinationAPI {
  setupEndpoints(app: Express) {
    // Health check with role information
    app.get('/health', async (req, res) => {
      res.json({
        status: 'healthy',
        role: this.config.role,
        instanceId: this.instanceId,
        partitions: await this.getAssignedPartitions(),
        metrics: await this.getMetrics()
      });
    });
    
    // Coordination endpoints
    app.post('/coordinate/rebalance', async (req, res) => {
      const result = await this.rebalancePartitions();
      res.json(result);
    });
    
    app.get('/coordinate/assignments', async (req, res) => {
      const assignments = await this.getPartitionAssignments();
      res.json(assignments);
    });
    
    // Sync endpoint for configuration updates
    app.post('/sync/config', async (req, res) => {
      await this.updateConfiguration(req.body);
      res.json({ status: 'updated' });
    });
  }
}

7. Event-Driven Coordination

Current State: No event system.

Proposed Enhancement:

// src/events/distributedEvents.ts
export class DistributedEventBus {
  private subscribers: Map<string, Set<EventHandler>> = new Map();
  
  async emit(event: BrainyEvent) {
    // Local handlers
    const handlers = this.subscribers.get(event.type) || new Set();
    for (const handler of handlers) {
      await handler(event);
    }
    
    // Remote broadcast (pluggable backends)
    await this.broadcast(event);
  }
  
  async broadcast(event: BrainyEvent) {
    // S3-based event log (simple, no additional deps)
    if (this.config.broadcastMethod === 's3') {
      await this.s3.putObject({
        Bucket: this.bucket,
        Key: `_brainy/events/${Date.now()}-${event.type}`,
        Body: JSON.stringify(event)
      });
    }
    
    // WebSocket broadcast
    if (this.config.broadcastMethod === 'websocket') {
      this.ws.broadcast(JSON.stringify(event));
    }
    
    // Cloud Pub/Sub
    if (this.config.broadcastMethod === 'pubsub') {
      await this.pubsub.topic('brainy-events').publish(event);
    }
  }
  
  subscribe(eventType: string, handler: EventHandler) {
    if (!this.subscribers.has(eventType)) {
      this.subscribers.set(eventType, new Set());
    }
    this.subscribers.get(eventType).add(handler);
  }
}

8. Configuration Synchronization

Current State: Local configuration only.

Proposed Enhancement:

// src/config/distributedConfig.ts
export class DistributedConfigManager {
  private configCache: ConfigCache;
  private configVersion: number = 0;
  
  async loadConfig(): Promise<BrainyConfig> {
    // Try S3 first for shared config
    try {
      const s3Config = await this.loadFromS3();
      this.configVersion = s3Config.version;
      return this.mergeWithLocal(s3Config);
    } catch (err) {
      // Fall back to local config
      return this.loadLocalConfig();
    }
  }
  
  async watchConfigChanges() {
    // Poll S3 for config updates
    setInterval(async () => {
      const latestVersion = await this.getConfigVersion();
      if (latestVersion > this.configVersion) {
        const newConfig = await this.loadFromS3();
        await this.applyConfigUpdate(newConfig);
        this.configVersion = latestVersion;
      }
    }, 10000);  // Check every 10 seconds
  }
  
  private async applyConfigUpdate(config: BrainyConfig) {
    // Hot-reload configuration without restart
    if (config.caching) {
      this.cacheManager.updateStrategy(config.caching);
    }
    if (config.partitioning) {
      await this.partitioner.reconfigure(config.partitioning);
    }
    // Emit event for other components
    this.eventBus.emit({
      type: 'config_updated',
      payload: config
    });
  }
}

Implementation Priority

Phase 1: Essential Features (Week 1-2)

1. Operational Modes - Enable read-only/write-only modes
2. S3 Coordination - Basic locking and manifest management
3. Configuration Sync - Shared configuration from S3

Phase 2: Coordination (Week 3-4)

4. Partition Assignment - Distributed partition management
5. Event System - Basic event broadcasting
6. Health Endpoints - Monitoring and coordination APIs

Phase 3: Advanced Features (Week 5-6)

7. Write-Ahead Log - Consistency and recovery
8. Advanced Coordination - WebSocket/Pub-Sub integration
9. Auto-rebalancing - Dynamic partition redistribution

Backwards Compatibility

All enhancements should be optional and backwards compatible:

// Default behavior remains unchanged
const brainy = new BrainyData({ /* existing config */ });

// Opt-in to distributed features
const distributedBrainy = new BrainyData({
  /* existing config */,
  distributed: {
    enabled: true,
    role: 'reader',
    // ... distributed options
  }
});

Testing Strategy

Unit Tests

• Test each operational mode independently
• Mock S3 coordination operations
• Test configuration synchronization

Integration Tests

• Multi-instance coordination tests
• Partition assignment and rebalancing
• Consistency under concurrent operations

Load Tests

• Simulate millions of vectors
• Test read/write separation at scale
• Measure coordination overhead

Performance Impact

Expected performance characteristics:

• Read-only instances: 10-20% faster due to optimized caching
• Write-only instances: 30-40% higher throughput with batching
• Coordination overhead: <100ms for most operations
• Configuration sync: <1s propagation delay

Conclusion

These enhancements would transform Brainy into a truly distributed vector database system capable of handling large-scale deployments with specialized instances. The modular design ensures that simpler use cases remain unaffected while enabling sophisticated distributed architectures when needed.