update publisher tests with better stability and performance

Bojan131 · Bojan131 · commit ad2363ced133 · 2026-01-09T15:31:41.000+01:00
diff --git a/apps/agent/.gitignore b/apps/agent/.gitignore
@@ -54,3 +54,4 @@ ragas-results.json
 logs
 storage
 docker-compose.knowledge-manager.yml
+DKG_Node_Publisher_Tests.xml
diff --git a/apps/agent/playwright.config.js b/apps/agent/playwright.config.js
@@ -7,7 +7,7 @@ const { defineConfig, devices } = require("@playwright/test");
 module.exports = defineConfig({
   testDir: "./tests/e2e",
   testMatch: "**/*.spec.js",
-  retries: 2,
+  retries: 2, // Retry failed tests 2 times (can be overridden per test file)
   workers: 1,
   timeout: 0, // No timeout per test (infinite polling for publisher plugin)
   globalTimeout: 0, // No global timeout
@@ -44,7 +44,7 @@ module.exports = defineConfig({
     /* Base URL to use in actions like `await page.goto('/')`. */
     baseURL: "http://localhost:8081",
     browserName: "chromium",
-    headless: true,
+    headless: false,
     actionTimeout: 0, // No timeout for actions (infinite polling)
     launchOptions: {
       slowMo: 1500,
diff --git a/apps/agent/tests/e2e/spec/publisherPluginTest.spec.js b/apps/agent/tests/e2e/spec/publisherPluginTest.spec.js
@@ -1,22 +1,100 @@
+/**
+ * Publisher Plugin Performance Test
+ * 
+ * Tests 10 knowledge asset publications with comprehensive metrics.
+ * 
+ * REQUIRED ENVIRONMENT VARIABLES (add to .env.testing.testnet.local):
+ * - PUBLISHING_TIMEOUT_MINUTES=5      # Timeout per publish (5 min for testing, 15 min production)
+ * - HEALTH_CHECK_INTERVAL_MS=10000    # Check every 10 seconds (60s production)
+ * - REDIS_HOST=localhost
+ * - DKGP_DATABASE_URL=mysql://root:@127.0.0.1:3306/dkg_publisher_db
+ * 
+ * See PUBLISHER_TEST_SETUP.md for detailed configuration instructions.
+ * 
+ * Test Duration:
+ * - Best case: 20-50 minutes (all succeed quickly)
+ * - Worst case: 50 minutes (all timeout at 5 min with maxAttempts=1)
+ */
+
 const { expect } = require("@playwright/test");
 const { test } = require("@playwright/test");
 const { Base } = require("../utils/base");
 const { LoginPage } = require("../pages/loginPage");
 const { ChatbotPage } = require("../pages/chatbotPage");
 const dotenv = require("dotenv");
 const path = require("path");
+const mysql = require("mysql2/promise");
+const Redis = require("ioredis");
 
 let base;
 let loginPage;
 let chatbotPage;
 
+// Disable retries for this test file (we test 10 assets in one run)
+test.describe.configure({ retries: 0 });
+
 function loadEnvFile(envFile) {
   dotenv.config({
     path: path.join(__dirname, "..", envFile),
     override: true,
   });
 }
 
+/**
+ * Clean up Redis and MySQL before test suite starts
+ * Ensures each test run starts with a fresh state
+ */
+test.beforeAll(async () => {
+  console.log("\n🧹 Cleaning up test environment...");
+
+  try {
+    // Clear Redis
+    const redis = new Redis({
+      host: process.env.REDIS_HOST || "localhost",
+      port: parseInt(process.env.REDIS_PORT || "6379"),
+      retryStrategy: () => null, // Don't retry on failure
+    });
+
+    await redis.flushall();
+    console.log("✅ Redis cleared");
+    await redis.quit();
+  } catch (error) {
+    console.warn("⚠️  Redis cleanup failed (may not be running):", error.message);
+  }
+
+  try {
+    // Clear MySQL
+    const dbUrl = process.env.DKGP_DATABASE_URL || "mysql://root:@127.0.0.1:3306/dkg_publisher_db";
+    const match = dbUrl.match(/mysql:\/\/([^:]+):([^@]*)@([^:]+):(\d+)\/(.+)/);
+    
+    if (!match) {
+      throw new Error("Invalid database URL format");
+    }
+
+    const [, user, password, host, port, database] = match;
+    
+    const connection = await mysql.createConnection({
+      host,
+      port: parseInt(port),
+      user,
+      password: password || undefined,
+      database,
+    });
+
+    await connection.execute("DELETE FROM publishing_attempts");
+    await connection.execute("DELETE FROM assets");
+    await connection.execute("UPDATE wallets SET is_locked = 0, locked_by = NULL, locked_at = NULL");
+    await connection.execute("ALTER TABLE assets AUTO_INCREMENT = 1");
+    
+    console.log("✅ Database cleared (assets, attempts, unlocked wallets)");
+    await connection.end();
+  } catch (error) {
+    console.warn("⚠️  Database cleanup failed:", error.message);
+  }
+
+  console.log("🚀 Test environment ready - starting with clean slate\n");
+});
+
 test.beforeEach(async ({ page }) => {
   base = new Base(page);
   await base.goToWebsite();
@@ -63,8 +141,9 @@ test("Publisher Plugin - Performance Test: Publish 10 Assets with Metrics", asyn
     const dateCreated = new Date().toISOString();
     const contentId = `urn:publisher-test:asset:${timestamp}-${i}`;
     
-    const publishMessage = `Register this Knowledge Asset as PRIVATE using the knowledge-asset-publish tool:
+    const publishMessage = `Register this Knowledge Asset using the knowledge-asset-publish tool with these settings:
 
+Content:
 {
   "@context": {
     "@vocab": "https://schema.org/"
@@ -77,7 +156,9 @@ test("Publisher Plugin - Performance Test: Publish 10 Assets with Metrics", asyn
   "testNumber": ${i}
 }
 
-Privacy: private
+Settings:
+- privacy: private
+- maxAttempts: 1
 `;
 
     const testStart = Date.now();
@@ -92,21 +173,35 @@ Privacy: private
       // Send publish request
       await chatbotPage.sendMessage(publishMessage);
 
-      // Wait for "Asset registered" response and check for "queued" confirmation (no timeout)
-      await page.waitForFunction(
-        () => {
-          const messages = Array.from(
-            document.querySelectorAll('[data-testid="chat-message-text"]'),
-          );
-          return messages.some((msg) => {
-            const text = msg.textContent || "";
-            return /queued for publishing/i.test(text) || 
-                   /Asset registered for publishing/i.test(text);
-          });
-        },
-        {},
-        { timeout: 0 },
-      );
+      // Wait for "Asset registered" response and check for "queued" confirmation
+      try {
+        await page.waitForFunction(
+          () => {
+            const messages = Array.from(
+              document.querySelectorAll('[data-testid="chat-message-text"]'),
+            );
+            return messages.some((msg) => {
+              const text = msg.textContent || "";
+              return /queued for publishing/i.test(text) || 
+                     /Asset registered for publishing/i.test(text) ||
+                     /registered.*publishing/i.test(text);
+            });
+          },
+          {},
+          { timeout: 30000 }, // 30 second timeout
+        );
+      } catch (timeoutError) {
+        // Log all messages to see what we actually got
+        const allMessages = await page.$$eval(
+          '[data-testid="chat-message-text"]',
+          (elements) => elements.map((el) => el.textContent || "")
+        );
+        console.log(`   ❌ Timeout waiting for queued confirmation. Messages received:`);
+        allMessages.forEach((msg, idx) => {
+          console.log(`      ${idx + 1}. ${msg.substring(0, 200)}`);
+        });
+        throw new Error(`Asset not queued - timeout waiting for confirmation. Last message: ${allMessages[allMessages.length - 1]?.substring(0, 200)}`);
+      }
 
       console.log(`   ✅ Asset queued with Content ID: ${contentId}`);
       console.log(`   ⏳ Polling status via chatbot until published or failed (no timeout)...`);
@@ -125,19 +220,8 @@ Privacy: private
           // Ask chatbot for status
           const statusQuery = `What is the status of ${contentId}?`;
           
-          // Add timeout wrapper for sendMessage to prevent indefinite hangs
-          const sendPromise = chatbotPage.sendMessage(statusQuery);
-          const sendTimeout = new Promise((_, reject) => 
-            setTimeout(() => reject(new Error('sendMessage timeout after 30s')), 30000)
-          );
-          await Promise.race([sendPromise, sendTimeout]);
-          
-          // Add timeout wrapper for waitForResponse to prevent indefinite hangs
-          const responsePromise = chatbotPage.waitForResponse(statusQuery);
-          const responseTimeout = new Promise((_, reject) => 
-            setTimeout(() => reject(new Error('waitForResponse timeout after 60s')), 60000)
-          );
-          await Promise.race([responsePromise, responseTimeout]);
+          await chatbotPage.sendMessage(statusQuery);
+          await chatbotPage.waitForResponse(statusQuery);
           
           // Get the last message from chatbot
           const messages = await page.getByTestId("chat-message-text").all();
diff --git a/packages/plugin-dkg-publisher/src/mcp/tools.ts b/packages/plugin-dkg-publisher/src/mcp/tools.ts
@@ -24,6 +24,9 @@ export function registerMcpTools(mcp: any, serviceContainer: ServiceContainer |
           })
           .optional(),
         privacy: z.enum(["private", "public"]).optional(),
+        maxAttempts: z.number().min(1).max(10).optional(),
+        epochs: z.number().optional(),
+        priority: z.number().min(1).max(100).optional(),
       },
     },
     async (input: any, req: any) => {
@@ -33,11 +36,18 @@ export function registerMcpTools(mcp: any, serviceContainer: ServiceContainer |
 
       const assetService = serviceContainer.get<AssetService>("assetService");
 
+      // Try to read maxAttempts from content.publishOptions first (for backward compatibility)
+      const maxAttemptsFromContent = input.content?.publishOptions?.maxAttempts;
+      const maxAttempts = input.maxAttempts || maxAttemptsFromContent || 3;
+
       const assetInput = {
         content: input.content,
         metadata: input.metadata,
         publishOptions: {
           privacy: input.privacy || "private",
+          maxAttempts: maxAttempts,
+          epochs: input.epochs,
+          priority: input.priority,
         },
       };
 
diff --git a/packages/plugin-dkg-publisher/src/services/HealthMonitor.ts b/packages/plugin-dkg-publisher/src/services/HealthMonitor.ts
@@ -25,13 +25,21 @@ export class HealthMonitor extends EventEmitter {
   private healthCheckInterval: NodeJS.Timeout | null = null;
   private isRunning: boolean = false;
   private queueService?: QueueService;
+  private readonly checkIntervalMs: number;
+  private readonly publishingTimeoutMinutes: number;
+  private readonly assignedTimeoutMinutes: number;
 
   constructor(
     private db: Database,
     private assetService: AssetService,
     private walletService: WalletService,
   ) {
     super();
+    
+    // Configure timeouts from environment (for testing flexibility)
+    this.checkIntervalMs = parseInt(process.env.HEALTH_CHECK_INTERVAL_MS || "10000"); // Default 10s
+    this.publishingTimeoutMinutes = parseInt(process.env.PUBLISHING_TIMEOUT_MINUTES || "5"); // Default 5 min
+    this.assignedTimeoutMinutes = parseInt(process.env.ASSIGNED_TIMEOUT_MINUTES || "2"); // Default 2 min
   }
 
    /**
@@ -44,15 +52,16 @@ export class HealthMonitor extends EventEmitter {
   /**
    * Start health monitoring
    */
-  start(intervalMs: number = 60000): void {
+  start(intervalMs?: number): void {
     if (this.isRunning) {
       console.log("Health monitor already running");
       return;
     }
 
+    const checkInterval = intervalMs || this.checkIntervalMs;
     this.isRunning = true;
     console.log(
-      `🏥 Health monitor started (checking every ${intervalMs / 1000} seconds)`,
+      `🏥 Health monitor started (checking every ${checkInterval / 1000} seconds, publishing timeout: ${this.publishingTimeoutMinutes} min)`,
     );
 
     // Stuck asset recovery is now handled by QueuePoller
@@ -63,7 +72,7 @@ export class HealthMonitor extends EventEmitter {
     // Set up periodic checks
     this.healthCheckInterval = setInterval(async () => {
       await this.checkHealth();
-    }, intervalMs);
+    }, checkInterval);
   }
 
   // Stuck asset recovery moved to QueuePoller for centralized queue management
@@ -110,8 +119,8 @@ export class HealthMonitor extends EventEmitter {
    * Check for stuck assets and handle them
    */
   private async checkStuckAssets(): Promise<void> {
-    // Check stuck "assigned" assets (>5 minutes)
-    const stuckAssigned = await this.assetService.getStuckAssets("assigned", 5);
+    // Check stuck "assigned" assets
+    const stuckAssigned = await this.assetService.getStuckAssets("assigned", this.assignedTimeoutMinutes);
 
     for (const asset of stuckAssigned) {
       console.log(
@@ -127,7 +136,7 @@ export class HealthMonitor extends EventEmitter {
             status: "queued",
             assignedAt: null,
             lastError:
-              "Timeout: assigned to wallet but publishing never started within 5 minutes",
+              `Timeout: assigned to wallet but publishing never started within ${this.assignedTimeoutMinutes} minutes`,
           })
           .where(eq(assets.id, asset.id));
 
@@ -143,15 +152,15 @@ export class HealthMonitor extends EventEmitter {
       });
     }
 
-    // Check stuck "publishing" assets (>15 minutes) - increased timeout for long operations
+    // Check stuck "publishing" assets
     const stuckPublishing = await this.assetService.getStuckAssets(
       "publishing",
-      15,
+      this.publishingTimeoutMinutes,
     );
 
     for (const asset of stuckPublishing) {
       console.log(
-        `⚠️ Asset ${asset.id} stuck in publishing status (publishing for over 15 minutes), handling failure...`,
+        `⚠️ Asset ${asset.id} stuck in publishing status (publishing for over ${this.publishingTimeoutMinutes} minutes), handling failure...`,
       );
 
       // 1. Find the latest publishing attempt and mark it as failed
@@ -169,8 +178,8 @@ export class HealthMonitor extends EventEmitter {
           .set({
             status: "failed",
             errorType: "Timeout",
-            errorMessage: "Publishing exceeded 15 minute timeout",
-            durationSeconds: 900, // 15 minutes
+            errorMessage: `Publishing exceeded ${this.publishingTimeoutMinutes} minute timeout`,
+            durationSeconds: this.publishingTimeoutMinutes * 60,
           })
           .where(eq(publishingAttempts.id, attemptId));
 
@@ -194,7 +203,7 @@ export class HealthMonitor extends EventEmitter {
       // 3. Handle asset failure with retry logic (increments retryCount)
       await this.assetService.handleAssetFailure(
         asset.id,
-        "Timeout: publishing over 15 minutes"
+        `Timeout: publishing over ${this.publishingTimeoutMinutes} minutes`
       );
 
       // 4. Release wallet lock
diff --git a/packages/plugin-dkg-publisher/src/services/QueueService.ts b/packages/plugin-dkg-publisher/src/services/QueueService.ts
diff --git a/packages/plugin-dkg-publisher/src/services/index.ts b/packages/plugin-dkg-publisher/src/services/index.ts
