vodorod/parcer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'develop' into abort-fix

Browse Source
This commit is contained in:
Rohit
2025-06-12 22:28:00 +05:30
committed by GitHub
parent a23f2f7bcb 154243b1de
commit 65472e079e
25 changed files with 902 additions and 412 deletions
Download Patch File Download Diff File Expand all files Collapse all files

111
server/src/browser-management/classes/BrowserPool.ts
View File

@@ -14,7 +14,7 @@ interface BrowserPoolInfo {
/**
* The instance of remote browser.
*/
browser: RemoteBrowser,
browser: RemoteBrowser | null,
/**
* States if the browser's instance is being actively used.
* Helps to persist the progress on the frontend when the application has been reloaded.
@@ -31,6 +31,11 @@ interface BrowserPoolInfo {
* @default "recording"
*/
state: BrowserState,
/**
* The status of the browser instance.
* Can be "reserved", "initializing", "ready" or "failed".
*/
status?: "reserved" | "initializing" | "ready" | "failed",
}
/**
@@ -205,8 +210,18 @@ export class BrowserPool {
* @returns remote browser instance or undefined if it does not exist in the pool
*/
public getRemoteBrowser = (id: string): RemoteBrowser | undefined => {
logger.log('debug', `Remote browser with id: ${id} retrieved from the pool`);
return this.pool[id]?.browser;
const poolInfo = this.pool[id];
if (!poolInfo) {
return undefined;
}
// Return undefined for reserved slots (browser is null)
if (poolInfo.status === "reserved") {
logger.log('debug', `Browser ${id} is reserved but not yet ready`);
return undefined;
}
return poolInfo.browser || undefined;
};
/**
@@ -506,6 +521,29 @@ export class BrowserPool {
return browserIds.length > 0 ? browserIds[0] : null;
};
/**
* Checks if there are available browser slots for a user.
* Returns true if user has available slots AND none of their active browsers are in "recording" state.
* @param userId the user ID to check browser slots for
* @returns {boolean} true if user has available slots and no recording browsers, false otherwise
*/
public hasAvailableBrowserSlots = (userId: string, state?: BrowserState): boolean => {
const userBrowserIds = this.userToBrowserMap.get(userId) || [];
if (userBrowserIds.length >= 2) {
return false;
}
if (state === "recording") {
const hasBrowserInState = userBrowserIds.some(browserId =>
this.pool[browserId] && this.pool[browserId].state === "recording"
);
return !hasBrowserInState;
}
return true;
};
/**
* Returns the first active browser's instance id from the pool.
* If there is no active browser, it returns null.
@@ -524,4 +562,71 @@ export class BrowserPool {
// logger.log('warn', `No active browser in the pool`);
return null;
};
/**
* Reserves a browser slot immediately without creating the actual browser.
* This ensures slot counting is accurate for rapid successive requests.
*
* @param id browser ID to reserve
* @param userId user ID that owns this reservation
* @param state browser state ("recording" or "run")
* @returns true if slot was reserved, false if user has reached limit
*/
public reserveBrowserSlot = (id: string, userId: string, state: BrowserState = "run"): boolean => {
// Check if user has available slots first
if (!this.hasAvailableBrowserSlots(userId, state)) {
logger.log('debug', `Cannot reserve slot for user ${userId}: no available slots`);
return false;
}
// Reserve the slot with null browser
this.pool[id] = {
browser: null,
active: false,
userId,
state,
status: "reserved"
};
// Update the user-to-browser mapping
let userBrowserIds = this.userToBrowserMap.get(userId) || [];
if (!userBrowserIds.includes(id)) {
userBrowserIds.push(id);
this.userToBrowserMap.set(userId, userBrowserIds);
}
logger.log('info', `Reserved browser slot ${id} for user ${userId} in state ${state}`);
return true;
};
/**
* Upgrades a reserved slot to an actual browser instance.
*
* @param id browser ID that was previously reserved
* @param browser the actual RemoteBrowser instance
* @returns true if successful, false if slot wasn't reserved
*/
public upgradeBrowserSlot = (id: string, browser: RemoteBrowser): boolean => {
if (!this.pool[id] || this.pool[id].status !== "reserved") {
logger.log('warn', `Cannot upgrade browser ${id}: slot not reserved`);
return false;
}
this.pool[id].browser = browser;
this.pool[id].status = "ready";
logger.log('info', `Upgraded browser slot ${id} to ready state`);
return true;
};
/**
* Marks a reserved slot as failed and removes it.
*
* @param id browser ID to mark as failed
*/
public failBrowserSlot = (id: string): void => {
if (this.pool[id]) {
logger.log('info', `Marking browser slot ${id} as failed`);
this.deleteRemoteBrowser(id);
}
};
}

69
server/src/browser-management/classes/RemoteBrowser.ts
View File

@@ -25,6 +25,13 @@ const MEMORY_CONFIG = {
heapUsageThreshold: 0.7 // 70% (reduced threshold to react earlier)
};
const DEFAULT_VIEWPORT = {
width: 1280,
height: 720,
deviceScaleFactor: 1,
mobile: false
};
const SCREENCAST_CONFIG: {
format: "jpeg" | "png";
maxWidth: number;
@@ -32,13 +39,17 @@ const SCREENCAST_CONFIG: {
targetFPS: number;
compressionQuality: number;
maxQueueSize: number;
skipFrameThreshold: number,
enableAdaptiveQuality: boolean,
} = {
format: 'png',
maxWidth: 1280,
maxHeight: 720,
targetFPS: 15,
compressionQuality: 0.95,
maxQueueSize: 1
format: 'jpeg',
maxWidth: DEFAULT_VIEWPORT.width,
maxHeight: DEFAULT_VIEWPORT.height,
targetFPS: 30,
compressionQuality: 0.8,
maxQueueSize: 2,
skipFrameThreshold: 100,
enableAdaptiveQuality: true,
};
/**
@@ -125,6 +136,18 @@ export class RemoteBrowser {
this.generator = new WorkflowGenerator(socket, poolId);
}
private cleanupMemory(): void {
if (this.screenshotQueue.length > 10) {
this.screenshotQueue = this.screenshotQueue.slice(-3); // Keep only last 3
}
}
private setupMemoryCleanup(): void {
setInterval(() => {
this.cleanupMemory();
}, 30000); // Every 30 seconds
}
private initializeMemoryManagement(): void {
setInterval(() => {
const memoryUsage = process.memoryUsage();
@@ -412,6 +435,7 @@ export class RemoteBrowser {
}
}
this.setupMemoryCleanup();
// this.initializeMemoryManagement();
};
@@ -1399,7 +1423,7 @@ export class RemoteBrowser {
*/
private emitScreenshot = async (payload: Buffer, viewportSize?: { width: number, height: number }): Promise<void> => {
if (this.screenshotQueue.length > SCREENCAST_CONFIG.maxQueueSize) {
this.screenshotQueue = this.screenshotQueue.slice(-SCREENCAST_CONFIG.maxQueueSize);
this.screenshotQueue = this.screenshotQueue.slice(-1);
}
if (this.isProcessingScreenshot) {
@@ -1414,7 +1438,7 @@ export class RemoteBrowser {
try {
const optimizationPromise = this.optimizeScreenshot(payload);
const timeoutPromise = new Promise<Buffer>((resolve) => {
setTimeout(() => resolve(payload), 150);
setTimeout(() => resolve(payload), 100);
});
const optimizedScreenshot = await Promise.race([optimizationPromise, timeoutPromise]);
@@ -1423,10 +1447,12 @@ export class RemoteBrowser {
payload = null as any;
this.socket.emit('screencast', {
setImmediate(async () => {
this.socket.emit('screencast', {
image: dataWithMimeType,
userId: this.userId,
viewport: viewportSize || await this.currentPage?.viewportSize() || null
});
});
} catch (error) {
logger.error('Screenshot emission failed:', error);
@@ -1434,24 +1460,27 @@ export class RemoteBrowser {
const base64Data = payload.toString('base64');
const dataWithMimeType = `data:image/png;base64,${base64Data}`;
this.socket.emit('screencast', {
image: dataWithMimeType,
userId: this.userId,
viewport: viewportSize || await this.currentPage?.viewportSize() || null
setImmediate(async () => {
this.socket.emit('screencast', {
image: dataWithMimeType,
userId: this.userId,
viewport: viewportSize || await this.currentPage?.viewportSize() || null
});
});
} catch (e) {
logger.error('Fallback screenshot emission also failed:', e);
}
} finally {
this.isProcessingScreenshot = false;
if (this.screenshotQueue.length > 0) {
const nextScreenshot = this.screenshotQueue.shift();
if (nextScreenshot) {
setTimeout(() => {
this.emitScreenshot(nextScreenshot);
}, 1000 / SCREENCAST_CONFIG.targetFPS);
}
const nextScreenshot = this.screenshotQueue.shift();
if (nextScreenshot) {
const delay = this.screenshotQueue.length > 0 ? 16 : 33;
setTimeout(() => {
this.emitScreenshot(nextScreenshot);
}, delay);
}
}
}
};

126
server/src/browser-management/controller.ts
View File

@@ -54,20 +54,23 @@ export const initializeRemoteBrowserForRecording = (userId: string): string => {
* @category BrowserManagement-Controller
*/
export const createRemoteBrowserForRun = (userId: string): string => {
const id = uuid();
if (!userId) {
logger.log('error', 'createRemoteBrowserForRun: Missing required parameter userId');
throw new Error('userId is required');
}
const id = uuid();
const slotReserved = browserPool.reserveBrowserSlot(id, userId, "run");
if (!slotReserved) {
logger.log('warn', `Cannot create browser for user ${userId}: no available slots`);
throw new Error('User has reached maximum browser limit');
}
logger.log('info', `createRemoteBrowserForRun: Reserved slot ${id} for user ${userId}`);
initializeBrowserAsync(id, userId);
createSocketConnectionForRun(
io.of(id),
async (socket: Socket) => {
try {
const browserSession = new RemoteBrowser(socket, userId, id);
await browserSession.initialize(userId);
browserPool.addRemoteBrowser(id, browserSession, userId, false, "run");
socket.emit('ready-for-run');
} catch (error: any) {
logger.error(`Error initializing browser: ${error.message}`);
}
});
return id;
};
@@ -135,6 +138,19 @@ export const getActiveBrowserIdByState = (userId: string, state: "recording" | "
return browserPool.getActiveBrowserId(userId, state);
};
/**
* Checks if there are available browser slots for a user.
* Wrapper around {@link browserPool.hasAvailableBrowserSlots()} function.
* If state is provided, also checks that none of their active browsers are in that state.
* @param userId the user ID to check browser slots for
* @param state optional state to check - if provided, ensures no browser is in this state
* @returns {boolean} true if user has available slots (and no browsers in specified state if state is provided)
* @category BrowserManagement-Controller
*/
export const canCreateBrowserInState = (userId: string, state?: "recording" | "run"): boolean => {
return browserPool.hasAvailableBrowserSlots(userId, state);
};
/**
* Returns the url string from a remote browser if exists in the browser pool.
* @param id instance id of the remote browser
@@ -198,3 +214,87 @@ export const stopRunningInterpretation = async (userId: string) => {
logger.log('error', 'Cannot stop interpretation: No active browser or generator.');
}
};
const initializeBrowserAsync = async (id: string, userId: string) => {
try {
const namespace = io.of(id);
let clientConnected = false;
let connectionTimeout: NodeJS.Timeout;
const waitForConnection = new Promise<Socket | null>((resolve) => {
namespace.on('connection', (socket: Socket) => {
clientConnected = true;
clearTimeout(connectionTimeout);
logger.log('info', `Frontend connected to browser ${id} via socket ${socket.id}`);
resolve(socket);
});
connectionTimeout = setTimeout(() => {
if (!clientConnected) {
logger.log('warn', `No client connected to browser ${id} within timeout, proceeding with dummy socket`);
resolve(null);
}
}, 10000);
});
namespace.on('error', (error: any) => {
logger.log('error', `Socket namespace error for browser ${id}: ${error.message}`);
clearTimeout(connectionTimeout);
browserPool.failBrowserSlot(id);
});
const socket = await waitForConnection;
try {
let browserSession: RemoteBrowser;
if (socket) {
logger.log('info', `Using real socket for browser ${id}`);
browserSession = new RemoteBrowser(socket, userId, id);
} else {
logger.log('info', `Using dummy socket for browser ${id}`);
const dummySocket = {
emit: (event: string, data?: any) => {
logger.log('debug', `Browser ${id} dummy socket emitted ${event}:`, data);
},
on: () => {},
id: `dummy-${id}`,
} as any;
browserSession = new RemoteBrowser(dummySocket, userId, id);
}
await browserSession.initialize(userId);
const upgraded = browserPool.upgradeBrowserSlot(id, browserSession);
if (!upgraded) {
throw new Error('Failed to upgrade reserved browser slot');
}
if (socket) {
socket.emit('ready-for-run');
} else {
setTimeout(async () => {
try {
logger.log('info', `Starting execution for browser ${id} with dummy socket`);
} catch (error: any) {
logger.log('error', `Error executing run for browser ${id}: ${error.message}`);
}
}, 100);
}
logger.log('info', `Browser ${id} successfully initialized for run with ${socket ? 'real' : 'dummy'} socket`);
} catch (error: any) {
logger.log('error', `Error initializing browser ${id}: ${error.message}`);
browserPool.failBrowserSlot(id);
if (socket) {
socket.emit('error', { message: error.message });
}
}
} catch (error: any) {
logger.log('error', `Error setting up browser ${id}: ${error.message}`);
browserPool.failBrowserSlot(id);
}
};

203
server/src/pgboss-worker.ts
View File

@@ -82,102 +82,13 @@ function AddGeneratedFlags(workflow: WorkflowFile) {
return copy;
};
/**
* Function to reset browser state without creating a new browser
*/
async function resetBrowserState(browser: RemoteBrowser): Promise<boolean> {
try {
const currentPage = browser.getCurrentPage();
if (!currentPage) {
logger.log('error', 'No current page available to reset browser state');
return false;
}
// Navigate to blank page to reset state
await currentPage.goto('about:blank', { waitUntil: 'networkidle', timeout: 10000 });
// Clear browser storage
await currentPage.evaluate(() => {
try {
localStorage.clear();
sessionStorage.clear();
} catch (e) {
// Ignore errors in cleanup
}
});
// Clear cookies
const context = currentPage.context();
await context.clearCookies();
return true;
} catch (error) {
logger.log('error', `Failed to reset browser state`);
return false;
}
}
/**
* Modified checkAndProcessQueuedRun function - only changes browser reset logic
*/
async function checkAndProcessQueuedRun(userId: string, browserId: string): Promise<boolean> {
try {
// Find the oldest queued run for this specific browser
const queuedRun = await Run.findOne({
where: {
browserId: browserId,
runByUserId: userId,
status: 'queued'
},
order: [['startedAt', 'ASC']]
});
if (!queuedRun) {
logger.log('info', `No queued runs found for browser ${browserId}`);
return false;
}
// Reset the browser state before next run
const browser = browserPool.getRemoteBrowser(browserId);
if (browser) {
logger.log('info', `Resetting browser state for browser ${browserId} before next run`);
await resetBrowserState(browser);
}
// Update the queued run to running status
await queuedRun.update({
status: 'running',
log: 'Run started - using browser from previous run'
});
// Use user-specific queue
const userQueueName = `execute-run-user-${userId}`;
// Schedule the run execution
await pgBoss.createQueue(userQueueName);
const executeJobId = await pgBoss.send(userQueueName, {
userId: userId,
runId: queuedRun.runId,
browserId: browserId
});
logger.log('info', `Scheduled queued run ${queuedRun.runId} to use browser ${browserId}, job ID: ${executeJobId}`);
return true;
} catch (error: unknown) {
const errorMessage = error instanceof Error ? error.message : String(error);
logger.log('error', `Error checking for queued runs: ${errorMessage}`);
return false;
}
}
/**
* Modified processRunExecution function - only add browser reset
*/
async function processRunExecution(job: Job<ExecuteRunData>) {
try {
const data = job.data;
logger.log('info', `Processing run execution job for runId: ${data.runId}, browserId: ${data.browserId}`);
const BROWSER_INIT_TIMEOUT = 30000;
const data = job.data;
logger.log('info', `Processing run execution job for runId: ${data.runId}`);
try {
// Find the run
const run = await Run.findOne({ where: { runId: data.runId } });
if (!run) {
@@ -191,6 +102,11 @@ async function processRunExecution(job: Job<ExecuteRunData>) {
}
const plainRun = run.toJSON();
const browserId = data.browserId || plainRun.browserId;
if (!browserId) {
throw new Error(`No browser ID available for run ${data.runId}`);
}
// Find the recording
const recording = await Robot.findOne({ where: { 'recording_meta.id': plainRun.robotMetaId }, raw: true });
@@ -231,33 +147,47 @@ async function processRunExecution(job: Job<ExecuteRunData>) {
}
}
// Check for queued runs even if this one failed
await checkAndProcessQueuedRun(data.userId, data.browserId);
return { success: false };
}
logger.log('info', `Looking for browser ${browserId} for run ${data.runId}`);
// Get the browser and execute the run
const browser = browserPool.getRemoteBrowser(plainRun.browserId);
let currentPage = browser?.getCurrentPage();
let browser = browserPool.getRemoteBrowser(browserId);
const browserWaitStart = Date.now();
if (!browser || !currentPage) {
logger.log('error', `Browser or page not available for run ${data.runId}`);
// Even if this run failed, check for queued runs
await checkAndProcessQueuedRun(data.userId, data.browserId);
return { success: false };
while (!browser && (Date.now() - browserWaitStart) < BROWSER_INIT_TIMEOUT) {
logger.log('debug', `Browser ${browserId} not ready yet, waiting...`);
await new Promise(resolve => setTimeout(resolve, 1000));
browser = browserPool.getRemoteBrowser(browserId);
}
if (!browser) {
throw new Error(`Browser ${browserId} not found in pool after timeout`);
}
logger.log('info', `Browser ${browserId} found and ready for execution`);
try {
// Reset the browser state before executing this run
await resetBrowserState(browser);
const isRunAborted = async (): Promise<boolean> => {
const currentRun = await Run.findOne({ where: { runId: data.runId } });
return currentRun ? (currentRun.status === 'aborted' || currentRun.status === 'aborting') : false;
};
let currentPage = browser.getCurrentPage();
const pageWaitStart = Date.now();
while (!currentPage && (Date.now() - pageWaitStart) < 30000) {
logger.log('debug', `Page not ready for browser ${browserId}, waiting...`);
await new Promise(resolve => setTimeout(resolve, 1000));
currentPage = browser.getCurrentPage();
}
if (!currentPage) {
throw new Error(`No current page available for browser ${browserId} after timeout`);
}
logger.log('info', `Starting workflow execution for run ${data.runId}`);
// Execute the workflow
const workflow = AddGeneratedFlags(recording.recording);
@@ -271,12 +201,7 @@ async function processRunExecution(job: Job<ExecuteRunData>) {
if (await isRunAborted()) {
logger.log('info', `Run ${data.runId} was aborted during execution, not updating status`);
const queuedRunProcessed = await checkAndProcessQueuedRun(data.userId, plainRun.browserId);
if (!queuedRunProcessed) {
await destroyRemoteBrowser(plainRun.browserId, data.userId);
logger.log('info', `No queued runs found for browser ${plainRun.browserId}, browser destroyed`);
}
await destroyRemoteBrowser(plainRun.browserId, data.userId);
return { success: true };
}
@@ -415,14 +340,8 @@ async function processRunExecution(job: Job<ExecuteRunData>) {
finishedAt: new Date().toLocaleString()
});
// Check for and process queued runs before destroying the browser
const queuedRunProcessed = await checkAndProcessQueuedRun(data.userId, plainRun.browserId);
// Only destroy the browser if no queued run was found
if (!queuedRunProcessed) {
await destroyRemoteBrowser(plainRun.browserId, data.userId);
logger.log('info', `No queued runs found for browser ${plainRun.browserId}, browser destroyed`);
}
await destroyRemoteBrowser(plainRun.browserId, data.userId);
logger.log('info', `No queued runs found for browser ${plainRun.browserId}, browser destroyed`);
return { success: true };
} catch (executionError: any) {
@@ -477,18 +396,7 @@ async function processRunExecution(job: Job<ExecuteRunData>) {
logger.log('info', `Run ${data.runId} was aborted, not updating status to failed`);
}
// Check for queued runs before destroying the browser
const queuedRunProcessed = await checkAndProcessQueuedRun(data.userId, plainRun.browserId);
// Only destroy the browser if no queued run was found
if (!queuedRunProcessed) {
try {
await destroyRemoteBrowser(plainRun.browserId, data.userId);
logger.log('info', `No queued runs found for browser ${plainRun.browserId}, browser destroyed`);
} catch (cleanupError: any) {
logger.log('warn', `Failed to clean up browser for failed run ${data.runId}: ${cleanupError.message}`);
}
}
await destroyRemoteBrowser(plainRun.browserId, data.userId);
return { success: false };
}
@@ -607,23 +515,14 @@ async function abortRun(runId: string, userId: string): Promise<boolean> {
} catch (socketError) {
logger.log('warn', `Failed to emit run-aborted event: ${socketError}`);
}
let queuedRunProcessed = false;
try {
queuedRunProcessed = await checkAndProcessQueuedRun(userId, plainRun.browserId);
} catch (queueError) {
logger.log('warn', `Error checking queued runs: ${queueError}`);
}
if (!queuedRunProcessed) {
try {
await new Promise(resolve => setTimeout(resolve, 500));
await destroyRemoteBrowser(plainRun.browserId, userId);
logger.log('info', `Browser ${plainRun.browserId} destroyed successfully after abort`);
} catch (cleanupError) {
logger.log('warn', `Failed to clean up browser for aborted run ${runId}: ${cleanupError}`);
}
try {
await new Promise(resolve => setTimeout(resolve, 500));
await destroyRemoteBrowser(plainRun.browserId, userId);
logger.log('info', `Browser ${plainRun.browserId} destroyed successfully after abort`);
} catch (cleanupError) {
logger.log('warn', `Failed to clean up browser for aborted run ${runId}: ${cleanupError}`);
}
return true;

13
server/src/routes/record.ts
View File

@@ -11,6 +11,7 @@ import {
getRemoteBrowserCurrentTabs,
getActiveBrowserIdByState,
destroyRemoteBrowser,
canCreateBrowserInState,
} from '../browser-management/controller';
import { chromium } from 'playwright-extra';
import stealthPlugin from 'puppeteer-extra-plugin-stealth';
@@ -181,6 +182,18 @@ router.get('/active', requireSignIn, (req: AuthenticatedRequest, res) => {
return res.send(id);
});
/**
* GET endpoint for checking if the user can create a new remote browser.
*/
router.get('/can-create/:state', requireSignIn, (req: AuthenticatedRequest, res) => {
if (!req.user) {
return res.status(401).send('User not authenticated');
}
const state = req.params.state as "recording" | "run";
const canCreate = canCreateBrowserInState(req.user.id, state);
return res.json({ canCreate });
});
/**
* GET endpoint for getting the current url of the active remote browser.
*/

202
server/src/routes/storage.ts
View File

@@ -1,6 +1,6 @@
import { Router } from 'express';
import logger from "../logger";
import { createRemoteBrowserForRun, getActiveBrowserIdByState } from "../browser-management/controller";
import { createRemoteBrowserForRun, destroyRemoteBrowser, getActiveBrowserIdByState } from "../browser-management/controller";
import { chromium } from 'playwright-extra';
import stealthPlugin from 'puppeteer-extra-plugin-stealth';
import { browserPool } from "../server";
@@ -517,55 +517,118 @@ router.put('/runs/:id', requireSignIn, async (req: AuthenticatedRequest, res) =>
return res.status(401).send({ error: 'Unauthorized' });
}
const proxyConfig = await getDecryptedProxyConfig(req.user.id);
let proxyOptions: any = {};
if (proxyConfig.proxy_url) {
proxyOptions = {
server: proxyConfig.proxy_url,
...(proxyConfig.proxy_username && proxyConfig.proxy_password && {
username: proxyConfig.proxy_username,
password: proxyConfig.proxy_password,
}),
};
}
console.log(`Proxy config for run: ${JSON.stringify(proxyOptions)}`);
// Generate runId first
const runId = uuid();
// Check if user has reached browser limit
const userBrowserIds = browserPool.getAllBrowserIdsForUser(req.user.id);
const canCreateBrowser = userBrowserIds.length < 2;
const canCreateBrowser = await browserPool.hasAvailableBrowserSlots(req.user.id, "run");
if (canCreateBrowser) {
// User has available browser slots, create it directly
const id = createRemoteBrowserForRun(req.user.id);
let browserId: string;
try {
browserId = await createRemoteBrowserForRun(req.user.id);
if (!browserId || browserId.trim() === '') {
throw new Error('Failed to generate valid browser ID');
}
logger.log('info', `Created browser ${browserId} for run ${runId}`);
} catch (browserError: any) {
logger.log('error', `Failed to create browser: ${browserError.message}`);
return res.status(500).send({ error: 'Failed to create browser instance' });
}
const run = await Run.create({
status: 'running',
try {
await Run.create({
status: 'running',
name: recording.recording_meta.name,
robotId: recording.id,
robotMetaId: recording.recording_meta.id,
startedAt: new Date().toLocaleString(),
finishedAt: '',
browserId: browserId,
interpreterSettings: req.body,
log: '',
runId,
runByUserId: req.user.id,
serializableOutput: {},
binaryOutput: {},
});
logger.log('info', `Created run ${runId} with browser ${browserId}`);
} catch (dbError: any) {
logger.log('error', `Database error creating run: ${dbError.message}`);
try {
await destroyRemoteBrowser(browserId, req.user.id);
} catch (cleanupError: any) {
logger.log('warn', `Failed to cleanup browser after run creation failure: ${cleanupError.message}`);
}
return res.status(500).send({ error: 'Failed to create run record' });
}
try {
const userQueueName = `execute-run-user-${req.user.id}`;
await pgBoss.createQueue(userQueueName);
const jobId = await pgBoss.send(userQueueName, {
userId: req.user.id,
runId: runId,
browserId: browserId,
});
logger.log('info', `Queued run execution job with ID: ${jobId} for run: ${runId}`);
} catch (queueError: any) {
logger.log('error', `Failed to queue run execution: ${queueError.message}`);
try {
await Run.update({
status: 'failed',
finishedAt: new Date().toLocaleString(),
log: 'Failed to queue execution job'
}, { where: { runId: runId } });
await destroyRemoteBrowser(browserId, req.user.id);
} catch (cleanupError: any) {
logger.log('warn', `Failed to cleanup after queue error: ${cleanupError.message}`);
}
return res.status(503).send({ error: 'Unable to queue run, please try again later' });
}
return res.send({
browserId: browserId,
runId: runId,
robotMetaId: recording.recording_meta.id,
queued: false
});
} else {
const browserId = uuid();
await Run.create({
status: 'queued',
name: recording.recording_meta.name,
robotId: recording.id,
robotMetaId: recording.recording_meta.id,
startedAt: new Date().toLocaleString(),
finishedAt: '',
browserId: id,
browserId,
interpreterSettings: req.body,
log: '',
log: 'Run queued - waiting for available browser slot',
runId,
runByUserId: req.user.id,
serializableOutput: {},
binaryOutput: {},
});
const plainRun = run.toJSON();
return res.send({
browserId: id,
runId: plainRun.runId,
browserId: browserId,
runId: runId,
robotMetaId: recording.recording_meta.id,
queued: false
queued: true
});
} else {
const browserId = getActiveBrowserIdByState(req.user.id, "run")
@@ -607,8 +670,8 @@ router.put('/runs/:id', requireSignIn, async (req: AuthenticatedRequest, res) =>
}
} catch (e) {
const { message } = e as Error;
logger.log('info', `Error while creating a run with robot id: ${req.params.id} - ${message}`);
return res.send('');
logger.log('error', `Error while creating a run with robot id: ${req.params.id} - ${message}`);
return res.status(500).send({ error: 'Internal server error' });
}
});
@@ -945,5 +1008,74 @@ router.post('/runs/abort/:id', requireSignIn, async (req: AuthenticatedRequest,
logger.log('error', `Error aborting run ${req.params.id}: ${message}`);
return res.status(500).send({ error: 'Failed to abort run' });
}
});
async function processQueuedRuns() {
try {
const queuedRun = await Run.findOne({
where: { status: 'queued' },
order: [['startedAt', 'ASC']]
});
if (!queuedRun) return;
const userId = queuedRun.runByUserId;
const canCreateBrowser = await browserPool.hasAvailableBrowserSlots(userId, "run");
if (canCreateBrowser) {
logger.log('info', `Processing queued run ${queuedRun.runId} for user ${userId}`);
const recording = await Robot.findOne({
where: {
'recording_meta.id': queuedRun.robotMetaId
},
raw: true
});
if (!recording) {
await queuedRun.update({
status: 'failed',
finishedAt: new Date().toLocaleString(),
log: 'Recording not found'
});
return;
}
try {
const newBrowserId = await createRemoteBrowserForRun(userId);
logger.log('info', `Created and initialized browser ${newBrowserId} for queued run ${queuedRun.runId}`);
await queuedRun.update({
status: 'running',
browserId: newBrowserId,
log: 'Browser created and ready for execution'
});
const userQueueName = `execute-run-user-${userId}`;
await pgBoss.createQueue(userQueueName);
const jobId = await pgBoss.send(userQueueName, {
userId: userId,
runId: queuedRun.runId,
browserId: newBrowserId,
});
logger.log('info', `Queued execution for run ${queuedRun.runId} with ready browser ${newBrowserId}, job ID: ${jobId}`);
} catch (browserError: any) {
logger.log('error', `Failed to create browser for queued run: ${browserError.message}`);
await queuedRun.update({
status: 'failed',
finishedAt: new Date().toLocaleString(),
log: `Failed to create browser: ${browserError.message}`
});
}
}
} catch (error: any) {
logger.log('error', `Error processing queued runs: ${error.message}`);
}
}
);
export { processQueuedRuns };

5
server/src/server.ts
View File

@@ -20,6 +20,7 @@ import connectPgSimple from 'connect-pg-simple';
import pg from 'pg';
import session from 'express-session';
import Run from './models/Run';
import { processQueuedRuns } from './routes/storage';
const app = express();
app.use(cors({
@@ -163,6 +164,10 @@ app.use((req, res, next) => {
next();
});
setInterval(() => {
processQueuedRuns();
}, 5000);
server.listen(SERVER_PORT, '0.0.0.0', async () => {
try {
await connectDB();

78
server/src/worker.ts
View File

@@ -1,78 +0,0 @@
import { Queue, Worker } from 'bullmq';
import IORedis from 'ioredis';
import logger from './logger';
import { handleRunRecording } from "./workflow-management/scheduler";
import Robot from './models/Robot';
import { computeNextRun } from './utils/schedule';
const connection = new IORedis({
host: process.env.REDIS_HOST,
port: process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT, 10) : 6379,
maxRetriesPerRequest: null,
password: process.env.REDIS_PASSWORD ? process.env.REDIS_PASSWORD : undefined,
});
connection.on('connect', () => {
console.log('Connected to Redis!');
});
connection.on('error', (err) => {
console.error('Redis connection error:', err);
});
const workflowQueue = new Queue('workflow', { connection });
const worker = new Worker('workflow', async job => {
const { runId, userId, id } = job.data;
try {
const result = await handleRunRecording(id, userId);
return result;
} catch (error) {
logger.error('Error running workflow:', error);
throw error;
}
}, { connection });
worker.on('completed', async (job: any) => {
logger.log(`info`, `Job ${job.id} completed for ${job.data.runId}`);
const robot = await Robot.findOne({ where: { 'recording_meta.id': job.data.id } });
if (robot) {
// Update `lastRunAt` to the current time
const lastRunAt = new Date();
// Compute the next run date
if (robot.schedule && robot.schedule.cronExpression && robot.schedule.timezone) {
const nextRunAt = computeNextRun(robot.schedule.cronExpression, robot.schedule.timezone) || undefined;
await robot.update({
schedule: {
...robot.schedule,
lastRunAt,
nextRunAt,
},
});
} else {
logger.error('Robot schedule, cronExpression, or timezone is missing.');
}
}
});
worker.on('failed', async (job: any, err) => {
logger.log(`error`, `Job ${job.id} failed for ${job.data.runId}:`, err);
});
console.log('Worker is running...');
async function jobCounts() {
const jobCounts = await workflowQueue.getJobCounts();
}
jobCounts();
// We dont need this right now
// process.on('SIGINT', () => {
// console.log('Worker shutting down...');
// process.exit();
// });
export { workflowQueue, worker };