Skip to content

Commit

Permalink
Merge pull request #14 from helyOSFramework/cancel_issue
Browse files Browse the repository at this point in the history 
Resolve Race Condition Affecting Cancel Signal Release to Agent
  • Loading branch information
@cviolbarbosa
cviolbarbosa authored May 13, 2024
2 parents 29dce91 + df2d69f commit 9113591
Show file tree
Hide file tree
Showing 2 changed files with 59 additions and 42 deletions.
78 changes: 42 additions & 36 deletions helyos_server/src/event_handlers/rabbitmq_event_handlers/status_event_handler.js
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,48 +2,54 @@
const databaseServices = require('../../services/database/database_services.js');
const { logData} = require('../..//modules/systemlog');
const { inMemDB } = require('../../services/in_mem_database/mem_database_service.js');
const { MISSION_STATUS, ASSIGNMENT_STATUS } = require('../../modules/data_models.js');


/* Update the assignmnet if it is not already marked as 'completed' or 'succeeded' */
/* The data is updated in the assignment table and in the agent table (under work process clearance) */
function updateAgentMission(assignment, uuid=null) {
if (!assignment) return Promise.resolve();
const assignment_status_obj = assignment.assignment_status? assignment.assignment_status:assignment; //back compatibility
async function updateAgentMission(assignment, uuid = null) {
if (!assignment) return;
const assignment_status_obj = assignment.assignment_status ? assignment.assignment_status : assignment; //back compatibility
const assignmentId = assignment_status_obj.id;
let assignmentStatus = assignment_status_obj.status;
const assignmentResult = assignment_status_obj.result;


if (assignmentId){
const assignment_update = {'id':assignmentId, 'status':assignmentStatus, 'result':assignmentResult };

return databaseServices.assignments.get_byId(assignmentId,['status'])
.then( assm => {
if (assm && assm.status !== 'succeeded' && assm.status !== 'completed' ){
return databaseServices.assignments.update_byId(assignment_update.id, assignment_update)
.then(() => databaseServices.agents.get('uuid', uuid, ['id', 'wp_clearance']))
.then(agents => {
if (!uuid) return;
if (agents.length == 0) {
logData.addLog('agent', {uuid}, 'error', "agent does not exist");
return;
}

const agent = agents[0];
if (agent.wp_clearance) agent.wp_clearance['assignment_status'] = assignment_update; //backward compatibility

databaseServices.agents.update_byId(agent.id, {'wp_clearance': agent.wp_clearance, 'assignment': assignment})
})

} else {
if (assignment_update.status!=='succeeded' && assignment_update.status !=='completed') {
logData.addLog('agent', {'uuid':uuid}, 'warning', `agent is trying to change an assignment that is already completed`);
}
}
});
if (!assignmentId) return;
const assignmentStatus = assignment_status_obj.status;
const assignmentResult = assignment_status_obj.result;

const assmUpdate = { 'id': assignmentId, 'status': assignmentStatus, 'result': assignmentResult };
const currentAssm = await databaseServices.assignments.get_byId(assignmentId, ['status', 'work_process_id']);

if (currentAssm &&
[ASSIGNMENT_STATUS.SUCCEEDED, ASSIGNMENT_STATUS.COMPLETED, ASSIGNMENT_STATUS.FAILED].includes(currentAssm.status)) {
logData.addLog('agent', { 'uuid': uuid }, 'warning', `agent tried to change the status of an assignment that is already ${currentAssm.status}`);
return;
}
return Promise.resolve();
}

if ([ASSIGNMENT_STATUS.CANCELED, ASSIGNMENT_STATUS.ABORTED, ASSIGNMENT_STATUS.FAILED].includes(assmUpdate.status)) {
logData.addLog('agent', { 'uuid': uuid }, 'info', `agent has marked the assignment ${assignmentId} as ${assmUpdate.status}`);
return await databaseServices.assignments.update_byId(assignmentId, assmUpdate);
}

await databaseServices.assignments.updateByConditions({
'assignments.id': assignmentId,
'work_processes.id': currentAssm.work_process_id,
'work_processes.status__in': [ MISSION_STATUS.EXECUTING,
MISSION_STATUS.DISPATCHED,
MISSION_STATUS.CALCULATING,
MISSION_STATUS.CANCELING,
MISSION_STATUS.FAILED]
}, assmUpdate);

if (uuid) {
const agents = await databaseServices.agents.get('uuid', uuid, ['id', 'wp_clearance']);
if (agents.length > 0) {
const agent = agents[0];
if (agent.wp_clearance) agent.wp_clearance['assignment_status'] = assmUpdate; //backward compatibility
return await databaseServices.agents.update_byId(agent.id, { 'wp_clearance': agent.wp_clearance, 'assignment': assignment });
} else {
logData.addLog('agent', { uuid }, 'error', "agent does not exist");
}
}
}



Expand Down
23 changes: 17 additions & 6 deletions helyos_server/src/services/database/postg_access_layer.js
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10,14 +10,24 @@ const { Client } = require('pg');
* @param {Object} conditions - key-value pairs
* @returns
*/
const parseConditions = (conditions) => {
const parseConditions = (tableName, conditions) => {
let names = [], values = [], masks = [];
let null_conditions = [], in_conditions = [];
if (!conditions || Object.keys(conditions).length == 0) {
return this.list();
}


let fromTableStatements = [];
Object.keys(conditions).forEach((key, idx) => {

if (key.includes('.')) {
const [table, field] = key.split('.');
if ( !tableName.includes(table) && !fromTableStatements.includes(table)) {
fromTableStatements.push(table);
}
}

if (key.endsWith('__in')) {
if (!Array.isArray(conditions[key])) {
throw new Error(`Invalid value for ${key}. Expected an array.`);
Expand All @@ -39,10 +49,11 @@ const parseConditions = (conditions) => {

names = names.join(',');
masks = masks.join(',');
fromTableStatements = fromTableStatements.length > 0 ? ' FROM ' + fromTableStatements.join(', ') : '';
null_conditions = null_conditions.join(' ');
in_conditions = in_conditions.join(' ');

return { names, values, masks, null_conditions, in_conditions };
return { names, values, masks, null_conditions, in_conditions, fromTableStatements };
}

class DatabaseLayer {
Expand Down Expand Up @@ -136,7 +147,7 @@ class DatabaseLayer {
else
selColNames = items.join(',');

const {names, values, masks, null_conditions, in_conditions} = parseConditions(conditions);
const {names, values, masks, null_conditions, in_conditions, fromTableStatements} = parseConditions(this.table, conditions);
const colNames = names, colValues = values, valueMasks = masks;

if (orderBy)
Expand Down Expand Up @@ -198,7 +209,7 @@ class DatabaseLayer {
return Promise.resolve({})
}

const {names, values, masks, null_conditions, in_conditions} = parseConditions(conditions);
const {names, values, masks, null_conditions, in_conditions, fromTableStatements} = parseConditions(this.table, conditions);
const condColNames = names, condColValues = values, condValueMasks = masks;

let colNames = [], colValues = [...condColValues], valueMasks = [];
Expand All @@ -223,9 +234,9 @@ class DatabaseLayer {


if (Object.keys(conditions).length > 1) {
queryText = 'UPDATE ' + this.table + ' SET ' + patchString + ' WHERE (' + condColNames + ') = (' + condValueMasks + ')' + null_conditions + in_conditions ;
queryText = 'UPDATE ' + this.table + ' SET ' + patchString + fromTableStatements + ' WHERE (' + condColNames + ') = (' + condValueMasks + ')' + null_conditions + in_conditions ;
} else {
queryText = 'UPDATE ' + this.table + ' SET ' + patchString + ' WHERE ' + condColNames + ' = ' + condValueMasks + null_conditions + in_conditions;
queryText = 'UPDATE ' + this.table + ' SET ' + patchString + fromTableStatements + ' WHERE ' + condColNames + ' = ' + condValueMasks + null_conditions + in_conditions;
}

const _client = useShortTimeClient ? this.shortTimeClient : this.client;
Expand Down

0 comments on commit 9113591

Please sign in to comment.