resubmit_limitbigint default100,

postpone interval,

max_run_timeinterval,

canceledboolean default false,

submit_timetimestamp with time zone default now()

);

CREATEINDEXat_jobs_submitted_node_at_idxon at_jobs_submitted (node, at);

AS'MODULE_PATHNAME','resubmit'

LANGUAGE C IMMUTABLE;

CREATEFUNCTIONcancel_job(job_idbigint) RETURNSbooleanAS

$BODY$

DECLARE

s_countint;

EXECUTE'SELECT count(*) FROM at_jobs_submitted WHERE owner = session_user AND id = $1' INTO s_count USING job_id;

IF s_count>0 THEN

UPDATE at_jobs_submittedSET canceled= trueWHERE"id"= job_id;

WITH moved_rowsAS (DELETEfrom ONLY at_jobs_submitted aWHEREa.id= job_id RETURNING a.*)INSERT INTO at_jobs_doneSELECT*,NULLas start_time, falseas status,'job was canceled'as reasonFROM moved_rows;

RETURN true;

ELSE

EXECUTE'SELECT count(*) FROM at_jobs_process WHERE owner = session_user AND id = $1' INTO s_count USING job_id;

IF s_count>0 THEN

UPDATE at_jobs_processSET canceled= trueWHERE"id"= job_id;

RETURN true;

END IF;

END IF;

RETURN false;

END

$BODY$

LANGUAGE plpgsql

SECURITY DEFINERset search_pathFROM CURRENT;

CREATEFUNCTIONsubmit_job(

querytext,

paramstext[] defaultNULL,

id, node, name, comments, atas run_after,

do_sqlas query, params, depends_on, executoras run_as, attempt,

resubmit_limit, postponeas max_wait_interval,

max_run_timeas max_duration, submit_time,

max_run_timeas max_duration, submit_time, canceled,

start_time, statusas is_success, reasonas error, done_time,

'done'::job_at_status_t status

FROMschedule.at_jobs_donewhere owner=session_user

id, node, name, comments, atas run_after,

do_sqlas query, params, depends_on, executoras run_as, attempt,

resubmit_limit, postponeas max_wait_interval,

max_run_timeas max_duration, submit_time, start_time,

max_run_timeas max_duration, submit_time,canceled,start_time,

NULLas is_success,NULLas error,NULLas done_time,

'processing'::job_at_status_t status

FROM ONLYschedule.at_jobs_processwhere owner=session_user

id, node, name, comments, atas run_after,

do_sqlas query, params, depends_on, executoras run_as, attempt,

resubmit_limit, postponeas max_wait_interval,

max_run_timeas max_duration, submit_time,

max_run_timeas max_duration, submit_time,canceled,

NULLas start_time,NULLas is_success,NULLas error,

NULLas done_time,

'submitted'::job_at_status_t status

Oidargtypes[2]= {TEXTOID,INT4OID };

Datumvalues[2];

constchar*get_job_sql="select id, at, last_start_available, max_run_time, executor from ONLY at_jobs_submitted s where ((not exists ( select * from ONLY at_jobs_submitted s2 where s2.id = any(s.depends_on)) AND not exists ( select * from ONLY at_jobs_process p where p.id = any(s.depends_on)) AND s.depends_on is NOT NULL and s.at IS NULL) OR ( s.at IS NOT NULL AND at <= 'now' and (last_start_available is NULL OR last_start_available > 'now'))) and node = $1 order by at, submit_time limit $2";

constchar*get_job_sql="select id, at, last_start_available, max_run_time, executor from ONLY at_jobs_submitted s where ((not exists ( select * from ONLY at_jobs_submitted s2 where s2.id = any(s.depends_on)) AND not exists ( select * from ONLY at_jobs_process p where p.id = any(s.depends_on)) AND s.depends_on is NOT NULL and s.at IS NULL) OR ( s.at IS NOT NULL AND at <= 'now' and (last_start_available is NULL OR last_start_available > 'now'))) and node = $1and not canceledorder by at, submit_time limit $2";

*is_error=*n=0;

START_SPI_SNAP();

intret;

constchar*sql="WITH moved_rows AS (DELETE from ONLY at_jobs_process a WHERE a.id = $1 RETURNING a.*) INSERT INTO at_jobs_submitted SELECT id, node, name, comments, $2, do_sql, params, depends_on, executor, owner, last_start_available, attempt +1 , resubmit_limit, postpone, max_run_time, submit_time FROM moved_rows";

values[0]=Int32GetDatum(j->cron_id);

values[1]=TimestampTzGetDatum(next);

ret=SPI_execute_with_args(sql,2,argtypes,values,NULL, false,0);

if(select_count_with_args("SELECT count(*) FROM at_jobs_process WHERE NOT canceled and id = $1",1,argtypes,values,NULL))

{

ret=SPI_execute_with_args(sql,2,argtypes,values,NULL, false,0);

}

{

return-2;

}

returnret>0 ?1:ret;

}

{

if(item->job->type==AtJob)

{

resubmit_at_job(item->job,shm_data->next_time);

if(resubmit_at_job(item->job,shm_data->next_time)==-2)

{

set_job_error(item->job,"was canceled while processing");

move_job_to_log(item->job, false, true);

}

}

{