diff --git a/Huly.md b/Huly.md new file mode 100644 index 0000000000..b7460d5473 --- /dev/null +++ b/Huly.md @@ -0,0 +1,74 @@ +# Huly on Network example + +## Building Huly on top of Huly Network + +Huly could be managed by following set of container kinds, `session`, `query`, `transactor`. + +- session -> a map/reduce/find executor for queries and transactions from client. +- query -> a DB query engine, execute `find` requests from session and pass them to DB, allow to search for all data per region. Should have access to tables of account -> workspace mapping for security. +- transactor -> modification archestrator for all edit operations, do them one by one. + +```mermaid +flowchart + Endpoint -.->| + connect + session/user1 + |HulyNetwork[Huly Network] + + Endpoint <-->|find,tx| parsonal-ws:user1 + + parsonal-ws:user1 -..->|get-workspace info| DatalakeDB + + parsonal-ws:user1 -..->|find| query:europe + + parsonal-ws:user1 -..->|event's| Endpoint + + query:europe -..->|resp| parsonal-ws:user1 + + parsonal-ws:user1 -..->|response chunks| Endpoint + + parsonal-ws:user1 -..->|tx| transactor:ws1 + + transactor:ws1 -..->|event's| HulyPulse + transactor:ws1 -..->|event's| parsonal-ws:user1 + + HulyPulse <--> Client + + Client <--> Endpoint + + query:europe -..->|"update"| QueryDB + transactor:ws1 -..->|update| DatalakeDB + + transactor:ws1 -..->|txes| Kafka[Output Queue] + + Kafka -..-> Indexer[Structure + + Fulltext Index] + + Indexer -..-> QueryDB + + Indexer -..->|indexed tx| HulyPulse + + Indexer -..->|indexed tx| parsonal-ws:user1 + + Kafka -..-> AsyncTriggers + + AsyncTriggers -..->|find| query:europe + + AsyncTriggers -..->|derived txes| transactor:ws1 + + InputQueue -->|txes| transactor:ws1 + + Services[Services + Github/Telegram/Translate] -..-> InputQueue + + Kafka -..-> Services + + Services -..-> query:europe + + QueryDB@{shape: database} + InputQueue@{shape: database} + DatalakeDB@{shape: database} + Kafka@{shape: database} + parsonal-ws:user1@{ shape: h-cyl} + +``` diff --git a/docs/HA_STATELESS_CONTAINERS.md b/docs/HA_STATELESS_CONTAINERS.md new file mode 100644 index 0000000000..a1da12a608 --- /dev/null +++ b/docs/HA_STATELESS_CONTAINERS.md @@ -0,0 +1,328 @@ +# High Availability (HA) Stateless Container Support + +This feature enables Huly Network to support stateless containers with automatic failover capabilities, allowing you to build highly available services that must ensure only one instance is active at any given time. + +## Overview + +The stateless container feature allows multiple agents to register pre-existing containers with the same UUID. The network will automatically: + +1. **Choose the first agent** that registers a container UUID +2. **Reject other agents** attempting to register the same UUID +3. **Enable automatic failover** when the active container is closed/terminated +4. **Allow standby agents to re-register** and take over when the primary fails + +This provides a simple yet effective HA mechanism for services that require leader election or single-instance guarantees. + +## Key Concepts + +### Stateless Containers + +Unlike regular containers that are created on-demand by agents, stateless containers are: + +- **Pre-existing**: The container instance already exists before registration +- **HA-aware**: Multiple agents can have the same container UUID ready to register +- **Failover-capable**: When removed, standby agents automatically attempt to take over + +### Registration Flow + +``` +Agent 1 (Primary) Network Agent 2 (Standby) + | | | + |-- Register UUID-001 ----->| | + |<-- Accepted ---------------| | + | |<--- Register UUID-001 -----| + | |---- Rejected (duplicate) ->| + | | | + | (container terminates) | | + |-- Unregister UUID-001 --->| | + | |-- Event: Removed --------->| + | | | + | |<--- Re-register UUID-001 --| + | |---- Accepted ------------->| +``` + +## API Reference + +### AgentImpl Methods + +#### `addStatelessContainer(uuid, kind, endpoint, container)` + +Add a stateless container to the agent for registration. + +**Parameters:** + +- `uuid: ContainerUuid` - The container UUID (must be consistent across HA agents) +- `kind: ContainerKind` - The container type/kind +- `endpoint: ContainerEndpointRef` - The endpoint reference for this container +- `container: Container` - The actual container instance + +**Example:** + +```typescript +const agent = new AgentImpl('my-agent-id', containerFactories) + +// Add a stateless container +agent.addStatelessContainer( + 'service-001' as ContainerUuid, + 'my-service' as ContainerKind, + 'service://host/service-001' as ContainerEndpointRef, + myServiceContainer +) + +// Register the agent - network will accept or reject the stateless container +await client.register(agent) +``` + +#### `removeStatelessContainer(uuid)` + +Remove a stateless container from tracking (called when rejected by network). + +**Parameters:** + +- `uuid: ContainerUuid` - The container UUID to remove + +### Network Behavior + +The network's `register()` method now: + +1. **Checks for UUID conflicts**: If a container UUID already exists and is owned by a different agent, the new registration is rejected +2. **Returns containers to shutdown**: The response includes UUIDs that should be terminated +3. **Broadcasts events**: Container removal events trigger standby agents to attempt re-registration + +## Usage Example + +### Basic HA Setup + +```typescript +import { AgentImpl, containerUuid, TickManagerImpl } from '@hcengineering/network-core' +import { createNetworkClient, NetworkAgentServer } from '@hcengineering/network-client' + +// Shared service UUID across all HA instances +const SHARED_SERVICE_UUID = 'my-ha-service-001' as ContainerUuid + +// Create Agent 1 (Primary) +const agent1 = new AgentImpl('agent-1', containerFactories) +agent1.addStatelessContainer( + SHARED_SERVICE_UUID, + 'ha-service', + 'service://agent-1/service-001', + primaryServiceContainer +) + +// Create Agent 2 (Standby) +const agent2 = new AgentImpl('agent-2', containerFactories) +agent2.addStatelessContainer( + SHARED_SERVICE_UUID, + 'ha-service', + 'service://agent-2/service-001', + standbyServiceContainer +) + +// Connect to network +const client = createNetworkClient('localhost:3737') +await client.waitConnection() + +// Register both agents +await client.register(agent1) // Will be accepted +await client.register(agent2) // Will be rejected for SHARED_SERVICE_UUID + +// When agent1's container is terminated, agent2 will automatically take over +``` + +### Monitoring and Failover + +```typescript +// Listen for container events to monitor failover +client.onUpdate(async (event) => { + for (const containerEvent of event.containers) { + if (containerEvent.container.uuid === SHARED_SERVICE_UUID) { + switch (containerEvent.event) { + case NetworkEventKind.added: + console.log('Container registered:', containerEvent.container.agentId) + break + case NetworkEventKind.removed: + console.log('Container removed - failover in progress') + break + case NetworkEventKind.updated: + console.log('Container updated') + break + } + } + } +}) + +// Simulate failover by terminating primary container +await agent1.terminate(SHARED_SERVICE_UUID) +// Agent2 will automatically re-register and take over (after ~100ms delay) +``` + +## Implementation Details + +### Agent Side + +1. **Stateless containers are tracked separately** in `AgentImpl.statelessContainers` map +2. **Included in registration** when `list()` is called +3. **Activated on success** - moved to active containers if accepted by network +4. **Terminated on rejection** - removed from tracking if network rejects + +### Network Side + +1. **First-wins policy** - The first agent to register a UUID owns it +2. **Rejection mechanism** - Returns UUIDs to shutdown for conflicting registrations +3. **Event broadcasting** - Sends removal events when containers are terminated +4. **Automatic cleanup** - Orphaned containers are tracked and cleaned up + +### Client Side + +1. **Event handling** - Monitors for container removal events +2. **Automatic re-registration** - Triggers re-registration when stateless container is removed +3. **Throttling** - Small delay (100ms) to prevent thundering herd +4. **Transparent to users** - Failover happens automatically + +## Use Cases + +### 1. Leader Election + +Implement leader election without external coordination services: + +```typescript +class LeaderService implements Container { + constructor(readonly clusterId: string) { + // Initialize leader-specific resources + } + + async request(operation: string, data?: any): Promise { + // Handle leader operations + return { isLeader: true, clusterId: this.clusterId } + } +} + +// All nodes register with same UUID, first one becomes leader +const leaderUuid = `cluster-${clusterId}-leader` as ContainerUuid +agent.addStatelessContainer(leaderUuid, 'leader', endpoint, new LeaderService(clusterId)) +``` + +### 2. Singleton Services + +Ensure only one instance of a service runs across the cluster: + +```typescript +class DatabaseMigrationService implements Container { + async request(operation: string): Promise { + if (operation === 'migrate') { + // Only one instance will run migrations + await this.runMigrations() + } + } +} + +const migrationUuid = 'db-migration-singleton' as ContainerUuid +agent.addStatelessContainer(migrationUuid, 'migration', endpoint, migrationService) +``` + +### 3. Active-Standby Databases + +Implement active-standby database pattern: + +```typescript +class DatabaseReplica implements Container { + private isActive = false + + async onActivation(): Promise { + // Promote standby to active + this.isActive = true + await this.promoteToMaster() + } + + async request(operation: string): Promise { + if (!this.isActive) { + throw new Error('Standby replica - read-only') + } + return await this.executeQuery(operation) + } +} +``` + +## Configuration + +### Failover Delay + +The default failover delay is 100ms. You can adjust this by modifying the timeout in `NetworkClientImpl.onEvent()`: + +```typescript +setTimeout(() => { + this.doRegister(agent).catch(...) +}, 100) // Adjust this value +``` + +### Container Cleanup + +Orphaned containers are automatically cleaned up based on the timeout settings. Configure via: + +```typescript +const tickManager = new TickManagerImpl(1) // 1 second tick +const network = new NetworkImpl(tickManager) +``` + +## Testing + +See the complete example in `examples/ha-stateless-container-example.ts`: + +```bash +# Start the network server +cd pods/network-pod +rushx dev + +# In another terminal, run the example +npx ts-node examples/ha-stateless-container-example.ts +``` + +## Limitations + +1. **No split-brain protection**: The network itself doesn't prevent split-brain scenarios if network partitions occur +2. **Eventually consistent**: There may be brief periods during failover where no instance is active +3. **No state transfer**: Stateless containers don't automatically transfer state between instances +4. **Single network dependency**: All agents must connect to the same Huly Network instance + +## Best Practices + +1. **Use meaningful UUIDs**: Make container UUIDs descriptive and consistent across deployments +2. **Implement health checks**: Containers should implement proper `ping()` methods +3. **Handle rejection gracefully**: Standby agents should be prepared to be rejected +4. **Monitor events**: Always subscribe to container events to track failover +5. **Test failover**: Regularly test failover scenarios in non-production environments +6. **Set appropriate timeouts**: Configure timeouts based on your SLA requirements + +## Troubleshooting + +### Problem: Standby agent not taking over + +**Solution**: Check that: + +- Both agents are properly connected to the network +- Container UUIDs match exactly +- Standby agent is properly listening for removal events +- Network event broadcasting is working + +### Problem: Both agents think they're active + +**Solution**: This indicates a split-brain scenario. Ensure: + +- All agents connect to the same network instance +- Network connections are stable +- No network partitions exist + +### Problem: Failover takes too long + +**Solution**: + +- Reduce the failover delay in `NetworkClientImpl.onEvent()` +- Implement faster health checks +- Use shorter network timeout values + +## See Also + +- [Container API](../packages/core/src/containers.ts) +- [Agent Implementation](../packages/core/src/agent.ts) +- [Network Implementation](../packages/core/src/network.ts) +- [Client Implementation](../packages/client/src/client.ts) diff --git a/docs/QUICKSTART_HA.md b/docs/QUICKSTART_HA.md new file mode 100644 index 0000000000..118f17cbcd --- /dev/null +++ b/docs/QUICKSTART_HA.md @@ -0,0 +1,160 @@ +# Quick Start: HA Stateless Containers + +## 5-Minute Guide + +### What is it? + +A feature that lets multiple agents compete to manage the same container UUID. The first agent wins, and others automatically take over if it fails. + +### When to use it? + +- You need leader election +- You want only one instance of a service running +- You need automatic failover +- You want HA without external coordination services + +### Basic Example + +```typescript +import { AgentImpl } from '@hcengineering/network-core' +import { createNetworkClient } from '@hcengineering/network-client' + +// 1. Create your container +class MyService implements Container { + constructor(readonly uuid: ContainerUuid) {} + + async request(operation: string): Promise { + return { status: 'active', uuid: this.uuid } + } + + async terminate(): Promise { + console.log('Service stopped') + } + + // ... other required methods +} + +// 2. Create agents (both will try to register same UUID) +const sharedUUID = 'my-service-001' as ContainerUuid + +// Agent 1 (Primary) +const agent1 = new AgentImpl('agent-1', {}) +const service1 = new MyService(sharedUUID) +agent1.addStatelessContainer( + sharedUUID, + 'my-service' as ContainerKind, + 'service://agent1/service-001' as ContainerEndpointRef, + service1 +) + +// Agent 2 (Standby) +const agent2 = new AgentImpl('agent-2', {}) +const service2 = new MyService(sharedUUID) +agent2.addStatelessContainer( + sharedUUID, + 'my-service' as ContainerKind, + 'service://agent2/service-001' as ContainerEndpointRef, + service2 +) + +// 3. Connect and register +const client = createNetworkClient('localhost:3737') +await client.waitConnection() + +await client.register(agent1) // ✅ Accepted +await client.register(agent2) // ❌ Rejected (agent1 already owns it) + +// 4. Failover happens automatically +await agent1.terminate(sharedUUID) // Agent1 stops +// After ~100ms, agent2 automatically takes over +``` + +### Key Methods + +```typescript +// Add a stateless container to agent +agent.addStatelessContainer(uuid, kind, endpoint, container) + +// Remove from tracking (if needed) +agent.removeStatelessContainer(uuid) + +// Register with network (handles conflicts automatically) +await client.register(agent) + +// Monitor for failover events +client.onUpdate(async (event) => { + for (const c of event.containers) { + if (c.event === NetworkEventKind.removed) { + console.log('Container removed - failover in progress') + } + } +}) +``` + +### What Happens? + +1. **Both agents register** → First one is accepted, second is rejected +2. **Rejected agent** → Terminates its container instance +3. **Active container fails** → Network broadcasts removal event +4. **Standby agents** → Automatically re-register (100ms delay) +5. **First standby wins** → Takes over as new active instance + +### Common Patterns + +#### Leader Election + +```typescript +const leaderId = `cluster-${clusterId}-leader` as ContainerUuid +agent.addStatelessContainer(leaderId, 'leader', endpoint, leaderService) +// First agent to register becomes leader +``` + +#### Singleton Service + +```typescript +const singletonId = 'migration-service' as ContainerUuid +agent.addStatelessContainer(singletonId, 'migration', endpoint, migrationService) +// Only one instance will run migrations +``` + +#### Active-Standby Database + +```typescript +const dbId = 'database-primary' as ContainerUuid +agent.addStatelessContainer(dbId, 'database', endpoint, databaseReplica) +// Primary serves writes, standby automatically promotes on failure +``` + +### Gotchas + +❌ **Don't** use different UUIDs on different agents (they won't compete) +✅ **Do** use the same UUID across all HA agents + +❌ **Don't** forget to handle termination of rejected containers +✅ **Do** let the agent.register() method handle it automatically + +❌ **Don't** expect instant failover (there's a ~100ms delay) +✅ **Do** design for eventual consistency + +### Testing + +```bash +# Terminal 1: Start network +cd pods/network-pod && rushx dev + +# Terminal 2: Run example +npx ts-node examples/ha-stateless-container-example.ts +``` + +### Next Steps + +- Read full docs: `docs/HA_STATELESS_CONTAINERS.md` +- See working example: `examples/ha-stateless-container-example.ts` +- Run tests: `cd packages/core && rushx test` + +### Questions? + +- How long is failover? ~100ms by default +- Can I have 3+ standbys? Yes, first to re-register wins +- Does state transfer? No, containers are stateless +- What about split-brain? No automatic protection (use network redundancy) diff --git a/examples/ha-stateless-container-example.ts b/examples/ha-stateless-container-example.ts new file mode 100644 index 0000000000..cbe71cf744 --- /dev/null +++ b/examples/ha-stateless-container-example.ts @@ -0,0 +1,218 @@ +/** + * Example: High Availability Stateless Container Registration + * + * This example demonstrates how to use stateless containers with multiple agents + * for HA (High Availability) scenarios. When multiple agents register the same + * container UUID, the network will choose the first one and reject others. + * + * When a container is closed/removed, standby agents will automatically attempt + * to re-register their instance, providing automatic failover. + * + * @example + * // Start the Huly Network server first: + * // cd pods/network-pod && rushx dev + * + * // Then run this example: + * // npx ts-node examples/ha-stateless-container-example.ts + */ + +import { AgentImpl, containerUuid as generateContainerUuid, TickManagerImpl } from '../packages/core/src' +import { createNetworkClient, NetworkAgentServer } from '../packages/client/src' +import type { + Container, + ContainerUuid, + ContainerKind, + ContainerEndpointRef, + ClientUuid, + GetOptions +} from '../packages/core/src' + +// Example stateless container implementation +class HAServiceContainer implements Container { + private connections = new Map Promise>() + + constructor ( + readonly uuid: ContainerUuid, + readonly serviceName: string + ) { + console.log(`[${serviceName}] Container ${uuid} created`) + } + + async request (operation: string, data?: any, clientId?: ClientUuid): Promise { + console.log(`[${this.serviceName}] Request: ${operation}`, data) + + switch (operation) { + case 'status': + return { + uuid: this.uuid, + serviceName: this.serviceName, + status: 'active', + timestamp: Date.now() + } + case 'shutdown': + console.log(`[${this.serviceName}] Shutdown requested`) + await this.terminate() + return { success: true } + default: + return { error: 'Unknown operation' } + } + } + + async ping (): Promise { + // Health check + } + + async terminate (): Promise { + console.log(`[${this.serviceName}] Terminating container ${this.uuid}`) + // Cleanup resources + } + + async connect (clientId: ClientUuid, broadcast: (data: any) => Promise): Promise { + this.connections.set(clientId, broadcast) + } + + async disconnect (clientId: ClientUuid): Promise { + this.connections.delete(clientId) + } +} + +/** + * Create an agent with stateless container support + */ +async function createHAAgent( + agentId: string, + agentName: string, + sharedServiceUuid: ContainerUuid, + port: number +): Promise<{ agent: AgentImpl, server: NetworkAgentServer }> { + const tickManager = new TickManagerImpl(1) + + // Create the agent with a container factory (for dynamic containers) + const agent = new AgentImpl( + agentId as any, + { + 'ha-service': async (options: GetOptions) => { + const uuid = options.uuid ?? generateContainerUuid() + const container = new HAServiceContainer(uuid, `${agentName}-dynamic`) + return { + uuid, + container, + endpoint: `ha-service://${agentName}/${uuid}` as ContainerEndpointRef + } + } + } as any + ) + + // Add a stateless container that already exists + // This simulates a pre-existing service that the agent wants to register + const statelessContainer = new HAServiceContainer(sharedServiceUuid, agentName) + agent.addStatelessContainer( + sharedServiceUuid, + 'ha-service' as ContainerKind, + `ha-service://${agentName}/${sharedServiceUuid}` as ContainerEndpointRef, + statelessContainer + ) + + console.log(`[${agentName}] Agent created with stateless container ${sharedServiceUuid}`) + + // Start the agent server + const server = new NetworkAgentServer(tickManager, 'localhost', '*', port) + await server.start(agent) + + return { agent, server } +} + +/** + * Main example demonstrating HA scenario + */ +async function main(): Promise { + console.log('=== HA Stateless Container Example ===\n') + + // Shared service UUID - both agents will try to register this + const sharedServiceUuid = 'service-leader-election-001' as ContainerUuid + + // Create network client + const client = createNetworkClient('localhost:3737') + await client.waitConnection(5000) + console.log('Connected to Huly Network\n') + + // Create two agents that will compete for the same container + console.log('Creating Agent 1 (Primary)...') + const { agent: agent1, server: server1 } = await createHAAgent( + 'ha-agent-1', + 'Primary', + sharedServiceUuid, + 3801 + ) + + console.log('Creating Agent 2 (Secondary)...') + const { agent: agent2, server: server2 } = await createHAAgent( + 'ha-agent-2', + 'Secondary', + sharedServiceUuid, + 3802 + ) + + // Register Agent 1 first - it should win + console.log('\n--- Registering Agent 1 ---') + await client.register(agent1) + await new Promise(resolve => setTimeout(resolve, 500)) + + // Register Agent 2 - it should be rejected for the shared container + console.log('\n--- Registering Agent 2 ---') + await client.register(agent2) + await new Promise(resolve => setTimeout(resolve, 500)) + + // Monitor container events + client.onUpdate(async (event: any) => { + console.log('\n>>> Network Event:') + for (const container of event.containers) { + console.log(` Container ${container.container.uuid}: ${['added', 'updated', 'removed'][container.event]}`) + } + }) + + // Verify which agent owns the container + console.log('\n--- Testing Container Access ---') + const containerRef = await client.get('ha-service' as ContainerKind, { + uuid: sharedServiceUuid + }) + const status = await containerRef.request('status') + console.log('Container status:', status) + + // Simulate failover: terminate the active container + console.log('\n--- Simulating Failover ---') + console.log('Shutting down primary container...') + await new Promise(resolve => setTimeout(resolve, 1000)) + + // Shutdown the primary agent's container + await agent1.terminate(sharedServiceUuid) + + // Wait for secondary to take over + console.log('Waiting for secondary agent to take over...') + await new Promise(resolve => setTimeout(resolve, 2000)) + + // Verify failover + try { + const newContainerRef = await client.get('ha-service' as ContainerKind, { + uuid: sharedServiceUuid + }) + const newStatus = await newContainerRef.request('status') + console.log('New container status after failover:', newStatus) + } catch (error) { + console.error('Failover verification failed:', error) + } + + // Cleanup + console.log('\n--- Cleanup ---') + await containerRef.close() + await client.close() + await server1.close() + await server2.close() + + console.log('\nExample completed!') +} + +// Run the example +main().catch(console.error) + +export { createHAAgent, HAServiceContainer } diff --git a/packages/client/src/client.ts b/packages/client/src/client.ts index 0815971835..99e0f78a69 100644 --- a/packages/client/src/client.ts +++ b/packages/client/src/client.ts @@ -179,6 +179,28 @@ export class NetworkClientImpl implements NetworkClient { // In case of container stopped, agent stopped or endpoint changed, we need to update direct connections to be re-established. await this.handleConnectionUpdates(event) + // Handle container removal for stateless containers - attempt to re-register + for (const containerEvent of event.containers) { + if (containerEvent.event === NetworkEventKind.removed) { + // Check if any of our agents have this container as stateless and need to re-register + for (const agentRecord of this._agents.values()) { + const agent = agentRecord.agent as any + const statelessContainers = agent.statelessContainers as Map | undefined + if (statelessContainers !== undefined && statelessContainers.has(containerEvent.container.uuid)) { + console.log( + `HA: Container ${containerEvent.container.uuid} removed, attempting to re-register from agent ${agent.uuid}` + ) + // Re-register this agent to attempt to claim the container + setTimeout(() => { + this.doRegister(agent).catch((err) => { + console.error(`Failed to re-register agent ${agent.uuid}:`, err) + }) + }, 100) // Small delay to avoid thundering herd + } + } + } + } + for (const listener of this.containerListeners.values()) { try { await listener(event) diff --git a/packages/core/src/__test__/ha-stateless.spec.ts b/packages/core/src/__test__/ha-stateless.spec.ts new file mode 100644 index 0000000000..7ba28d27a3 --- /dev/null +++ b/packages/core/src/__test__/ha-stateless.spec.ts @@ -0,0 +1,170 @@ +/** + * Test suite for HA Stateless Container feature + */ + +import { AgentImpl, containerUuid, NetworkImpl, TickManagerImpl } from '../index' +import type { Container, ContainerUuid, ContainerKind, ClientUuid, ContainerEndpointRef, GetOptions } from '../index' + +class MockHAContainer implements Container { + terminated = false + + constructor (readonly uuid: ContainerUuid, readonly name: string) {} + + async request (operation: string, data?: any): Promise { + return { uuid: this.uuid, name: this.name, operation, data } + } + + async ping (): Promise {} + + async terminate (): Promise { + this.terminated = true + } + + connect (clientId: ClientUuid, broadcast: (data: any) => Promise): void {} + disconnect (clientId: ClientUuid): void {} +} + +describe('HA Stateless Containers', () => { + const agentId1 = 'agent-1' as any + const agentId2 = 'agent-2' as any + const clientId1 = 'client-1' as any + const haServiceKind = 'ha-service' as ContainerKind + const sharedUuid = 'shared-container-001' as ContainerUuid + + let tickManager: TickManagerImpl + let network: NetworkImpl + + beforeEach(() => { + tickManager = new TickManagerImpl(1) + network = new NetworkImpl(tickManager) + }) + + test('first agent wins when multiple agents register same UUID', async () => { + // Create two agents with the same stateless container UUID + const agent1 = new AgentImpl(agentId1, { + [haServiceKind]: async (opt: GetOptions) => ({ + uuid: opt.uuid ?? containerUuid(), + container: new MockHAContainer(opt.uuid ?? containerUuid(), 'agent1-dynamic'), + endpoint: 'endpoint1' as ContainerEndpointRef + }) + }) + + const agent2 = new AgentImpl(agentId2, { + [haServiceKind]: async (opt: GetOptions) => ({ + uuid: opt.uuid ?? containerUuid(), + container: new MockHAContainer(opt.uuid ?? containerUuid(), 'agent2-dynamic'), + endpoint: 'endpoint2' as ContainerEndpointRef + }) + }) + + // Add stateless containers with same UUID + const container1 = new MockHAContainer(sharedUuid, 'agent1-stateless') + const container2 = new MockHAContainer(sharedUuid, 'agent2-stateless') + + agent1.addStatelessContainer(sharedUuid, haServiceKind, 'endpoint1' as ContainerEndpointRef, container1) + agent2.addStatelessContainer(sharedUuid, haServiceKind, 'endpoint2' as ContainerEndpointRef, container2) + + // Register agent1 first + await agent1.register(network) + const list1 = await agent1.list() + expect(list1.some(c => c.uuid === sharedUuid)).toBe(true) // Should be accepted + + // Register agent2 - should be rejected for the shared UUID + await agent2.register(network) + const list2 = await agent2.list() + // Container should have been removed from agent2's list after rejection + expect(list2.some(c => c.uuid === sharedUuid)).toBe(false) + + // Verify agent1 owns the container + const [uuid, endpoint] = await network.get(clientId1, haServiceKind, { uuid: sharedUuid }) + expect(uuid).toBe(sharedUuid) + expect(endpoint).toBe('endpoint1') + }) + + test('stateless container is activated on successful registration', async () => { + const agent = new AgentImpl(agentId1, { + [haServiceKind]: async (opt: GetOptions) => ({ + uuid: opt.uuid ?? containerUuid(), + container: new MockHAContainer(opt.uuid ?? containerUuid(), 'dynamic'), + endpoint: 'endpoint' as ContainerEndpointRef + }) + }) + + const statelessContainer = new MockHAContainer(sharedUuid, 'stateless') + agent.addStatelessContainer(sharedUuid, haServiceKind, 'endpoint' as ContainerEndpointRef, statelessContainer) + + // Before registration, container should be in stateless map + expect((agent as any).statelessContainers.has(sharedUuid)).toBe(true) + + await agent.register(network) + + // After successful registration, container should be activated (moved to active containers) + expect((agent as any).statelessContainers.has(sharedUuid)).toBe(false) + expect((agent as any)._byId.has(sharedUuid)).toBe(true) + }) + + test('stateless container is terminated when rejected', async () => { + const agent1 = new AgentImpl(agentId1, {}) + const agent2 = new AgentImpl(agentId2, {}) + + const container1 = new MockHAContainer(sharedUuid, 'agent1') + const container2 = new MockHAContainer(sharedUuid, 'agent2') + + agent1.addStatelessContainer(sharedUuid, haServiceKind, 'endpoint1' as ContainerEndpointRef, container1) + agent2.addStatelessContainer(sharedUuid, haServiceKind, 'endpoint2' as ContainerEndpointRef, container2) + + await agent1.register(network) + await agent2.register(network) + + // Container2 should have been terminated as it was rejected + expect(container2.terminated).toBe(true) + expect(container1.terminated).toBe(false) + }) + + test('stateless containers are included in list() call', async () => { + const agent = new AgentImpl(agentId1, {}) + + const statelessContainer = new MockHAContainer(sharedUuid, 'stateless') + agent.addStatelessContainer(sharedUuid, haServiceKind, 'endpoint' as ContainerEndpointRef, statelessContainer) + + const list = await agent.list() + expect(list).toHaveLength(1) + expect(list[0].uuid).toBe(sharedUuid) + expect(list[0].kind).toBe(haServiceKind) + }) + + test('getContainer works for both active and stateless containers', async () => { + const agent = new AgentImpl(agentId1, {}) + + const statelessContainer = new MockHAContainer(sharedUuid, 'stateless') + agent.addStatelessContainer(sharedUuid, haServiceKind, 'endpoint' as ContainerEndpointRef, statelessContainer) + + // Before activation, should find in stateless containers + const container = await agent.getContainer(sharedUuid) + expect(container).toBe(statelessContainer) + + // After activation + await agent.register(network) + const container2 = await agent.getContainer(sharedUuid) + expect(container2).toBe(statelessContainer) + }) + + test('same agent re-registering updates endpoint', async () => { + const agent = new AgentImpl(agentId1, {}) + + const container = new MockHAContainer(sharedUuid, 'agent1') + agent.addStatelessContainer(sharedUuid, haServiceKind, 'endpoint1' as ContainerEndpointRef, container) + + await agent.register(network) + + // Re-register with different endpoint + agent.removeStatelessContainer(sharedUuid) + agent.addStatelessContainer(sharedUuid, haServiceKind, 'endpoint2' as ContainerEndpointRef, container) + + await agent.register(network) + + // Should not be rejected, endpoint should be updated + const [, endpoint] = await network.get(clientId1, haServiceKind, { uuid: sharedUuid }) + expect(endpoint).toBe('endpoint2') + }) +}) diff --git a/packages/core/src/agent.ts b/packages/core/src/agent.ts index 427d41718a..0382748dad 100644 --- a/packages/core/src/agent.ts +++ b/packages/core/src/agent.ts @@ -39,11 +39,61 @@ export class AgentImpl implements NetworkAgent { endpoint?: AgentEndpointRef | undefined + // Stateless containers that need to be registered + private readonly statelessContainers = new Map() + constructor ( readonly uuid: AgentUuid, private readonly factory: Record ) {} + /** + * Add a stateless container to the agent. + * Stateless containers are pre-existing containers that the agent wants to register with the network. + * Used for HA scenarios where multiple agents may try to register the same container UUID. + */ + addStatelessContainer ( + uuid: ContainerUuid, + kind: ContainerKind, + endpoint: ContainerEndpointRef, + container: Container + ): void { + const record: ContainerRecordImpl = { + container, + uuid, + endpoint, + kind, + lastVisit: Date.now() + } + this.statelessContainers.set(uuid, record) + } + + /** + * Remove a stateless container from tracking. + * This is called when the network rejects our registration (another agent won). + */ + removeStatelessContainer (uuid: ContainerUuid): void { + this.statelessContainers.delete(uuid) + } + + /** + * Activate a stateless container after successful registration. + * Moves it from stateless tracking to active containers. + */ + private activateStatelessContainer (uuid: ContainerUuid): void { + const record = this.statelessContainers.get(uuid) + if (record !== undefined) { + this._byId.set(uuid, record) + const byKind = this._byKind.get(record.kind) + if (byKind !== undefined) { + byKind.set(uuid, record) + } else { + this._byKind.set(record.kind, new Map([[uuid, record]])) + } + this.statelessContainers.delete(uuid) + } + } + async register (network: Network): Promise { const cleanContainers = await network.register( { @@ -54,13 +104,23 @@ export class AgentImpl implements NetworkAgent { }, this ) + + // Activate stateless containers that were accepted + for (const record of this.statelessContainers.values()) { + if (!cleanContainers.includes(record.uuid)) { + this.activateStatelessContainer(record.uuid) + } + } + + // Terminate containers that network wants cleaned up for (const c of cleanContainers) { await this.terminate(c) } } async list (kind?: ContainerKind): Promise { - return Array.from(kind !== undefined ? (this._byKind.get(kind)?.values() ?? []) : this._byId.values()) + // Include both active and stateless containers + const active = Array.from(kind !== undefined ? (this._byKind.get(kind)?.values() ?? []) : this._byId.values()) .filter((it) => !(it instanceof Promise) && (kind === undefined || it.kind === kind)) .map((it) => ({ agentId: this.uuid, @@ -69,6 +129,18 @@ export class AgentImpl implements NetworkAgent { kind: it.kind, lastVisit: it.lastVisit })) + + const stateless = Array.from(this.statelessContainers.values()) + .filter((it) => kind === undefined || it.kind === kind) + .map((it) => ({ + agentId: this.uuid, + uuid: it.uuid, + endpoint: it.endpoint, + kind: it.kind, + lastVisit: it.lastVisit + })) + + return [...active, ...stateless] } get kinds (): ContainerKind[] { @@ -87,7 +159,12 @@ export class AgentImpl implements NetworkAgent { } async getContainer (uuid: ContainerUuid): Promise { - return this._byId.get(uuid)?.container + const active = this._byId.get(uuid) + if (active !== undefined) { + return active.container + } + // Check stateless containers too + return this.statelessContainers.get(uuid)?.container } async get (kind: ContainerKind, options: GetOptions): Promise<[ContainerUuid, ContainerEndpointRef]> { @@ -160,6 +237,14 @@ export class AgentImpl implements NetworkAgent { } else { await current.container.terminate() } + return + } + + // Also check stateless containers + const stateless = this.statelessContainers.get(uuid) + if (stateless !== undefined) { + this.statelessContainers.delete(uuid) + await stateless.container.terminate() } } diff --git a/packages/core/src/network.ts b/packages/core/src/network.ts index 438c6a2916..0dbd1c1c1c 100644 --- a/packages/core/src/network.ts +++ b/packages/core/src/network.ts @@ -174,16 +174,24 @@ export class NetworkImpl implements Network, NetworkWithClients { this._containers.set(containerImpl.record.uuid, containerImpl) agentRecord.containers.add(containerImpl.record.uuid) } else { + // Container already exists - HA scenario if (existingContainer.agent.id !== record.agentId) { - // Container already started on different agent, need to shutdown old one. + // Container already started on different agent + // The first agent wins, tell this agent to shutdown this container + console.log( + `HA: Container ${containerImpl.record.uuid} already owned by agent ${existingContainer.agent.id}, rejecting agent ${record.agentId}` + ) containersToShutdown.push(containerImpl.record.uuid) - } - - if (existingContainer.record.endpoint !== containerImpl.record.endpoint) { - containerEvent.containers.push({ - container: containerImpl.record, - event: NetworkEventKind.updated - }) + } else { + // Same agent re-registering - update endpoint if changed + if (existingContainer.record.endpoint !== containerImpl.record.endpoint) { + containerEvent.containers.push({ + container: containerImpl.record, + event: NetworkEventKind.updated + }) + existingContainer.record.endpoint = containerImpl.record.endpoint + existingContainer.endpoint = containerImpl.record.endpoint + } } } }