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

feat: map cleanup, process duration

Browse Source
This commit is contained in:
Rohit Rajan
2025-09-28 22:36:43 +05:30
parent 0f759170e7
commit 0aeb8ad6ca
2 changed files with 27 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
server/src/workflow-management/integrations/airtable.ts
View File

@@ -454,7 +454,10 @@ function isValidUrl(str: string): boolean {
} }
export const processAirtableUpdates = async () => { export const processAirtableUpdates = async () => {
while (true) { const maxProcessingTime = 60000;
const startTime = Date.now();
while (Date.now() - startTime < maxProcessingTime) {
let hasPendingTasks = false; let hasPendingTasks = false;
for (const runId in airtableUpdateTasks) { for (const runId in airtableUpdateTasks) {
@@ -467,7 +470,6 @@ export const processAirtableUpdates = async () => {
try { try {
await updateAirtable(task.robotId, task.runId); await updateAirtable(task.robotId, task.runId);
console.log(`Successfully updated Airtable for runId: ${runId}`); console.log(`Successfully updated Airtable for runId: ${runId}`);
airtableUpdateTasks[runId].status = 'completed';
delete airtableUpdateTasks[runId]; delete airtableUpdateTasks[runId];
} catch (error: any) { } catch (error: any) {
console.error(`Failed to update Airtable for run ${task.runId}:`, error); console.error(`Failed to update Airtable for run ${task.runId}:`, error);
@@ -476,11 +478,12 @@ export const processAirtableUpdates = async () => {
airtableUpdateTasks[runId].retries += 1; airtableUpdateTasks[runId].retries += 1;
console.log(`Retrying task for runId: ${runId}, attempt: ${task.retries + 1}`); console.log(`Retrying task for runId: ${runId}, attempt: ${task.retries + 1}`);
} else { } else {
airtableUpdateTasks[runId].status = 'failed'; console.log(`Max retries reached for runId: ${runId}. Removing task.`);
console.log(`Max retries reached for runId: ${runId}. Marking task as failed.`); delete airtableUpdateTasks[runId];
logger.log('error', `Permanent failure for run ${runId}: ${error.message}`);
} }
} }
} else if (task.status === 'completed' || task.status === 'failed') {
delete airtableUpdateTasks[runId];
} }
} }
@@ -492,4 +495,6 @@ export const processAirtableUpdates = async () => {
console.log('Waiting for 5 seconds before checking again...'); console.log('Waiting for 5 seconds before checking again...');
await new Promise(resolve => setTimeout(resolve, 5000)); await new Promise(resolve => setTimeout(resolve, 5000));
} }
console.log('Airtable processing completed or timed out');
}; };

15
server/src/workflow-management/integrations/gsheet.ts
View File

@@ -286,8 +286,12 @@ export async function writeDataToSheet(
} }
export const processGoogleSheetUpdates = async () => { export const processGoogleSheetUpdates = async () => {
while (true) { const maxProcessingTime = 60000;
const startTime = Date.now();
while (Date.now() - startTime < maxProcessingTime) {
let hasPendingTasks = false; let hasPendingTasks = false;
for (const runId in googleSheetUpdateTasks) { for (const runId in googleSheetUpdateTasks) {
const task = googleSheetUpdateTasks[runId]; const task = googleSheetUpdateTasks[runId];
console.log(`Processing task for runId: ${runId}, status: ${task.status}`); console.log(`Processing task for runId: ${runId}, status: ${task.status}`);
@@ -297,7 +301,6 @@ export const processGoogleSheetUpdates = async () => {
try { try {
await updateGoogleSheet(task.robotId, task.runId); await updateGoogleSheet(task.robotId, task.runId);
console.log(`Successfully updated Google Sheet for runId: ${runId}`); console.log(`Successfully updated Google Sheet for runId: ${runId}`);
googleSheetUpdateTasks[runId].status = 'completed';
delete googleSheetUpdateTasks[runId]; delete googleSheetUpdateTasks[runId];
} catch (error: any) { } catch (error: any) {
console.error(`Failed to update Google Sheets for run ${task.runId}:`, error); console.error(`Failed to update Google Sheets for run ${task.runId}:`, error);
@@ -305,10 +308,12 @@ export const processGoogleSheetUpdates = async () => {
googleSheetUpdateTasks[runId].retries += 1; googleSheetUpdateTasks[runId].retries += 1;
console.log(`Retrying task for runId: ${runId}, attempt: ${task.retries}`); console.log(`Retrying task for runId: ${runId}, attempt: ${task.retries}`);
} else { } else {
googleSheetUpdateTasks[runId].status = 'failed'; console.log(`Max retries reached for runId: ${runId}. Removing task.`);
console.log(`Max retries reached for runId: ${runId}. Marking task as failed.`); delete googleSheetUpdateTasks[runId];
} }
} }
} else if (task.status === 'completed' || task.status === 'failed') {
delete googleSheetUpdateTasks[runId];
} }
} }
@@ -320,4 +325,6 @@ export const processGoogleSheetUpdates = async () => {
console.log('Waiting for 5 seconds before checking again...'); console.log('Waiting for 5 seconds before checking again...');
await new Promise(resolve => setTimeout(resolve, 5000)); await new Promise(resolve => setTimeout(resolve, 5000));
} }
console.log('Google Sheets processing completed or timed out');
}; };