chore: fix retry

Author: coodos

infrastructure/web3-adapter/src/evault/evault.ts

@@ -17,6 +17,8 @@ const CONFIG = {
     MAX_RETRIES: 3,
     RETRY_DELAY: 1000, // 1 second base delay
     CONNECTION_POOL_SIZE: 10,
+    HEALTH_CHECK_TIMEOUT: 5000, // 5 seconds for health check
+    MAX_HEALTH_CHECK_FAILURES: 3, // Max consecutive failures before re-resolution
 } as const;
 
 const STORE_META_ENVELOPE = `
@@ -126,6 +128,10 @@ export class EVaultClient {
     private endpoints: Map<string, string> = new Map();
     private tokenInfo: TokenInfo | null = null;
     private isDisposed = false;
+    
+    // Health check tracking
+    private healthCheckFailures: Map<string, number> = new Map();
+    private lastHealthCheck: Map<string, number> = new Map();
 
     constructor(
         private registryUrl: string,
@@ -141,6 +147,8 @@ export class EVaultClient {
         this.isDisposed = true;
         this.clients.clear();
         this.endpoints.clear();
+        this.healthCheckFailures.clear();
+        this.lastHealthCheck.clear();
         this.tokenInfo = null;
     }
 
@@ -274,8 +282,15 @@ export class EVaultClient {
         if (this.clients.has(w3id)) {
             const client = this.clients.get(w3id)!;
             const endpoint = this.endpoints.get(w3id)!;
-            console.log('reusing existing client for w3id:', w3id, 'endpoint:', endpoint);
-            return client;
+            
+            // Check if the cached endpoint is still healthy
+            if (await this.isEndpointHealthy(w3id, endpoint)) {
+                console.log('reusing existing client for w3id:', w3id, 'endpoint:', endpoint);
+                return client;
+            } else {
+                console.log('cached endpoint is unhealthy, removing and re-resolving for w3id:', w3id);
+                this.removeCachedClient(w3id);
+            }
         }
 
         // Resolve endpoint for this specific w3id
@@ -294,10 +309,153 @@ export class EVaultClient {
         this.clients.set(w3id, client);
         this.endpoints.set(w3id, endpoint);
 
+        // Initialize health check tracking
+        this.healthCheckFailures.set(w3id, 0);
+        this.lastHealthCheck.set(w3id, Date.now());
+        
         console.log('created new client for w3id:', w3id, 'endpoint:', endpoint);
         return client;
     }
 
+    /**
+     * Check if a cached endpoint is still healthy
+     */
+    private async isEndpointHealthy(w3id: string, endpoint: string): Promise<boolean> {
+        try {
+            // Extract base URL from GraphQL endpoint
+            const baseUrl = endpoint.replace('/graphql', '');
+            
+            // Check if we should perform health check (avoid too frequent checks)
+            const now = Date.now();
+            const lastCheck = this.lastHealthCheck.get(w3id) || 0;
+            const timeSinceLastCheck = now - lastCheck;
+            
+            // Only check every 30 seconds to avoid performance impact
+            if (timeSinceLastCheck < 30000) {
+                return true; // Assume healthy if checked recently
+            }
+            
+            // Perform health check on the whois endpoint
+            const healthCheckUrl = `${baseUrl}/whois`;
+            console.log(`Health checking endpoint for ${w3id}: ${healthCheckUrl}`);
+            
+            const controller = new AbortController();
+            const timeoutId = setTimeout(() => controller.abort(), CONFIG.HEALTH_CHECK_TIMEOUT);
+            
+            const response = await fetch(healthCheckUrl, {
+                method: 'HEAD',
+                signal: controller.signal,
+            });
+            
+            clearTimeout(timeoutId);
+            
+            if (response.ok) {
+                // Reset failure count on success
+                this.healthCheckFailures.set(w3id, 0);
+                this.lastHealthCheck.set(w3id, now);
+                return true;
+            } else {
+                throw new Error(`Health check failed with status: ${response.status}`);
+            }
+            
+        } catch (error) {
+            console.log(`Health check failed for ${w3id}:`, error instanceof Error ? error.message : 'Unknown error');
+            
+            // Increment failure count
+            const currentFailures = this.healthCheckFailures.get(w3id) || 0;
+            const newFailures = currentFailures + 1;
+            this.healthCheckFailures.set(w3id, newFailures);
+            this.lastHealthCheck.set(w3id, Date.now());
+            
+            // If we've had too many consecutive failures, mark as unhealthy
+            if (newFailures >= CONFIG.MAX_HEALTH_CHECK_FAILURES) {
+                console.log(`Endpoint for ${w3id} marked as unhealthy after ${newFailures} consecutive failures`);
+                return false;
+            }
+            
+            // Still allow some failures before marking as unhealthy
+            return true;
+        }
+    }
+
+    /**
+     * Remove cached client and endpoint for a specific w3id
+     */
+    private removeCachedClient(w3id: string): void {
+        this.clients.delete(w3id);
+        this.endpoints.delete(w3id);
+        this.healthCheckFailures.delete(w3id);
+        this.lastHealthCheck.delete(w3id);
+        console.log(`Removed cached client for ${w3id}`);
+    }
+
+    /**
+     * Manually trigger a health check for a specific w3id
+     * Useful for testing or forcing re-resolution
+     */
+    public async forceHealthCheck(w3id: string): Promise<boolean> {
+        if (!this.clients.has(w3id)) {
+            console.log(`No cached client found for ${w3id}`);
+            return false;
+        }
+
+        const endpoint = this.endpoints.get(w3id);
+        if (!endpoint) {
+            console.log(`No cached endpoint found for ${w3id}`);
+            return false;
+        }
+
+        // Force health check by clearing last check time
+        this.lastHealthCheck.set(w3id, 0);
+        
+        const isHealthy = await this.isEndpointHealthy(w3id, endpoint);
+        
+        if (!isHealthy) {
+            console.log(`Forced health check failed for ${w3id}, removing cached client`);
+            this.removeCachedClient(w3id);
+        }
+        
+        return isHealthy;
+    }
+
+    /**
+     * Get health status for all cached endpoints
+     */
+    public getHealthStatus(): Record<string, {
+        endpoint: string;
+        failures: number;
+        lastCheck: number;
+        isHealthy: boolean;
+    }> {
+        const status: Record<string, any> = {};
+        
+        for (const [w3id, endpoint] of this.endpoints) {
+            const failures = this.healthCheckFailures.get(w3id) || 0;
+            const lastCheck = this.lastHealthCheck.get(w3id) || 0;
+            const isHealthy = failures < CONFIG.MAX_HEALTH_CHECK_FAILURES;
+            
+            status[w3id] = {
+                endpoint,
+                failures,
+                lastCheck,
+                isHealthy,
+            };
+        }
+        
+        return status;
+    }
+
+    /**
+     * Clear all cached clients (useful for testing or forcing fresh connections)
+     */
+    public clearCache(): void {
+        console.log('Clearing all cached clients and endpoints');
+        this.clients.clear();
+        this.endpoints.clear();
+        this.healthCheckFailures.clear();
+        this.lastHealthCheck.clear();
+    }
+
     async storeMetaEnvelope(envelope: MetaEnvelope): Promise<string> {
         return this.withRetry(async () => {
             const client = await this.ensureClient(envelope.w3id).catch(() => {

infrastructure/web3-adapter/tsconfig.json

@@ -2,7 +2,7 @@
     "compilerOptions": {
         "target": "ES2021",
         "module": "NodeNext",
-        "moduleResolution": "node",
+        "moduleResolution": "NodeNext",
         "lib": ["ES2020"],
         "declaration": true,
         "outDir": "./dist",

platforms/registry/package.json

@@ -15,10 +15,11 @@
         "migration:create": "npm run typeorm migration:create"
     },
     "dependencies": {
-        "@fastify/cors": "8.4.1",
+        "@fastify/cors": "^8.4.0",
+        "@kubernetes/client-node": "^0.20.0",
         "@fastify/jwt": "^7.2.3",
         "axios": "^1.6.7",
-        "dotenv": "^16.5.0",
+        "dotenv": "^16.4.5",
         "fastify": "^4.26.1",
         "jose": "^5.2.2",
         "pg": "^8.11.3",

platforms/registry/src/index.ts

@@ -4,6 +4,7 @@ import dotenv from "dotenv";
 import path from "path";
 import { AppDataSource } from "./config/database";
 import { VaultService } from "./services/VaultService";
+import { UriResolutionService } from "./services/UriResolutionService";
 import cors from "@fastify/cors";
 
 dotenv.config({ path: path.resolve(__dirname, "../../../.env") });
@@ -32,6 +33,9 @@ const initializeDatabase = async () => {
 // Initialize VaultService
 const vaultService = new VaultService(AppDataSource.getRepository("Vault"));
 
+// Initialize UriResolutionService for health checks and Kubernetes fallbacks
+const uriResolutionService = new UriResolutionService();
+
 // Middleware to check shared secret
 const checkSharedSecret = async (request: any, reply: any) => {
     const authHeader = request.headers.authorization;
@@ -138,10 +142,15 @@ server.get("/resolve", async (request, reply) => {
             return reply.status(404).send({ error: "Service not found" });
         }
 
+        // Resolve the URI with health check and Kubernetes fallback
+        const resolvedUri = await uriResolutionService.resolveUri(vault.uri);
+
         return {
             ename: vault.ename,
-            uri: vault.uri,
+            uri: resolvedUri,
             evault: vault.evault,
+            originalUri: vault.uri, // Include original URI for debugging
+            resolved: resolvedUri !== vault.uri, // Flag if URI was resolved
         };
     } catch (error) {
         server.log.error(error);
@@ -153,11 +162,22 @@ server.get("/resolve", async (request, reply) => {
 server.get("/list", async (request, reply) => {
     try {
         const vaults = await vaultService.findAll();
-        return vaults.map(vault => ({
-            ename: vault.ename,
-            uri: vault.uri,
-            evault: vault.evault,
-        }));
+        
+        // Resolve URIs for all vaults
+        const resolvedVaults = await Promise.all(
+            vaults.map(async (vault) => {
+                const resolvedUri = await uriResolutionService.resolveUri(vault.uri);
+                return {
+                    ename: vault.ename,
+                    uri: resolvedUri,
+                    evault: vault.evault,
+                    originalUri: vault.uri,
+                    resolved: resolvedUri !== vault.uri,
+                };
+            })
+        );
+        
+        return resolvedVaults;
     } catch (error) {
         server.log.error(error);
         reply.status(500).send({ error: "Failed to list vault entries" });