resubmit && resubmit_limit

Author: Vladimir Ershov

pgpro_scheduler--2.0.sql

@@ -17,6 +17,8 @@ CREATE TABLE at_jobs_submitted(
    executor text,
    owner text,
    last_start_available timestamp with time zone,
+   attempt bigint default 0,
+   resubmit_limit bigint default 100,
    postpone interval,
    max_run_time	interval,
    submit_time timestamp with time zone default now()
@@ -153,8 +155,8 @@ CREATE FUNCTION get_self_id()
   AS 'MODULE_PATHNAME', 'get_self_id'
   LANGUAGE C IMMUTABLE;
 
-CREATE FUNCTION resubmit()
-  RETURNS boolean 
+CREATE FUNCTION resubmit(run_after interval default NULL)
+  RETURNS bigint 
   AS 'MODULE_PATHNAME', 'resubmit'
   LANGUAGE C IMMUTABLE;
 
@@ -168,7 +170,8 @@ CREATE FUNCTION submit_job(
 	run_as text default NULL,
 	depends_on bigint[] default NULL,
 	name text default NULL,
-	comments text default NULL
+	comments text default NULL,
+	resubmit_limit bigint default 100
 ) RETURNS bigint AS
 $BODY$
 DECLARE
@@ -215,10 +218,12 @@ BEGIN
 
 	INSERT INTO at_jobs_submitted
 		(node, at, do_sql, owner, executor, name, comments, max_run_time,
-		 postpone, last_start_available, depends_on, params)
+		 postpone, last_start_available, depends_on, params,
+		 attempt, resubmit_limit)
 	VALUES
 		(node, run_after, query, session_user, executor,  name, comments,
-		 max_duration, max_wait_interval, last_avail, depends_on, params)
+		 max_duration, max_wait_interval, last_avail, depends_on, params,
+		 0, resubmit_limit)
 	RETURNING id INTO job_id;
 
 	RETURN job_id;

src/pgpro_scheduler.c

@@ -92,7 +92,7 @@ void reload_db_role_config(char *dbname)
 	CommitTransactionCommand();
 }
 
-TimestampTz timestamp_add_seconds(TimestampTz to, int add)
+TimestampTz timestamp_add_seconds(TimestampTz to, int64 add)
 {
 	if(to == 0) to = GetCurrentTimestamp();
 #ifdef HAVE_INT64_TIMESTAMP

src/pgpro_scheduler.h

@@ -34,7 +34,7 @@ int checkSchedulerNamespace(void);
 pid_t registerManagerWorker(schd_manager_t *man);
 
 void reload_db_role_config(char *dbname);
-TimestampTz timestamp_add_seconds(TimestampTz to, int add);
+TimestampTz timestamp_add_seconds(TimestampTz to, int64 add);
 char *make_date_from_timestamp(TimestampTz ts, bool hires);
 int get_integer_from_string(char *s, int start, int len);
 TimestampTz get_timestamp_from_string(char *str);

src/scheduler_executor.c

@@ -36,7 +36,7 @@ extern volatile sig_atomic_t got_sighup;
 extern volatile sig_atomic_t got_sigterm;
 
 static int64 current_job_id = -1;
-static int resubmit_current_job = 0;
+static int64 resubmit_current_job = 0;
 
 static void handle_sigterm(SIGNAL_ARGS);
 
@@ -193,7 +193,24 @@ void executor_worker_main(Datum arg)
 		{
 			STOP_SPI_SNAP();
 		}
-		status = SchdExecutorDone;
+		if(job->type == AtJob && resubmit_current_job > 0)
+		{
+			if(job->attempt >= job->resubmit_limit)
+			{
+				status = SchdExecutorError;
+				push_executor_error(&EE, "Cannot resubmit: limit reached (%ld)", job->resubmit_limit);
+				resubmit_current_job = 0;
+			}
+			else
+			{
+				status = SchdExecutorResubmit;
+			}
+		}
+		else
+		{
+			status = SchdExecutorDone;
+		}
+
 		SetConfigOption("schedule.transaction_state", "success", PGC_INTERNAL, PGC_S_SESSION);
 	}
 	if(job->next_time_statement)
@@ -218,6 +235,11 @@ void executor_worker_main(Datum arg)
 		set_shared_message(shared, &EE);
 	}
 	shared->status = status;
+	if(status == SchdExecutorResubmit)
+	{
+		shared->next_time = timestamp_add_seconds(0, resubmit_current_job);
+		resubmit_current_job = 0;
+	}
 
 	delete_worker_mem_ctx();
 	dsm_detach(seg);
@@ -481,9 +503,28 @@ PG_FUNCTION_INFO_V1(resubmit);
 Datum 
 resubmit(PG_FUNCTION_ARGS)
 {
+	Interval *interval;	
+
 	if(current_job_id == -1)
 	{
 		elog(ERROR, "There is no active job in progress");	
 	}
-	PG_RETURN_BOOL(true);
+	if(PG_ARGISNULL(0))
+	{
+		resubmit_current_job = 1;		
+		PG_RETURN_INT64(1);
+	}
+	interval = PG_GETARG_INTERVAL_P(0);
+#ifdef HAVE_INT64_TIMESTAMP 
+    resubmit_current_job = interval->time / 1000000.0;
+#else
+    resubmit_current_job = interval->time;
+#endif
+	resubmit_current_job +=
+		(DAYS_PER_YEAR * SECS_PER_DAY) * (interval->month / MONTHS_PER_YEAR);
+	resubmit_current_job +=
+		(DAYS_PER_MONTH * SECS_PER_DAY) * (interval->month % MONTHS_PER_YEAR);
+	resubmit_current_job += SECS_PER_DAY * interval->day;
+
+	PG_RETURN_INT64(resubmit_current_job);
 }

src/scheduler_executor.h

@@ -11,6 +11,7 @@ typedef enum {
 	SchdExecutorInit,
 	SchdExecutorWork,
 	SchdExecutorDone,
+	SchdExecutorResubmit,
 	SchdExecutorError
 } schd_executor_status_t;
 

src/scheduler_job.c

@@ -25,7 +25,7 @@ job_t *init_scheduler_job(job_t *j, unsigned char type)
 job_t *get_at_job(int cron_id, char *nodename, char **perror)
 {
 	job_t *j;
-	const char *sql = "select last_start_available, array_append('{}'::text[], do_sql)::text[], executor, postpone, max_run_time as time_limit, at, params, depends_on from ONLY at_jobs_process where node = $1 and id = $2";
+	const char *sql = "select last_start_available, array_append('{}'::text[], do_sql)::text[], executor, postpone, max_run_time as time_limit, at, params, depends_on, attempt, resubmit_limit from ONLY at_jobs_process where node = $1 and id = $2";
 	Oid argtypes[2] = { TEXTOID, INT4OID};
 	Datum args[2];
 	int ret;
@@ -69,6 +69,8 @@ job_t *get_at_job(int cron_id, char *nodename, char **perror)
 		j->start_at = get_timestamp_from_spi(0, 6, 0);
 		j->sql_params = get_textarray_from_spi(0, 7, &j->sql_params_n);
 		j->depends_on = get_int64array_from_spi(0, 8, &j->depends_on_n);
+		j->attempt = get_int64_from_spi(0, 9, 0);
+		j->resubmit_limit = get_int64_from_spi(0, 10, 0);
 
 		STOP_SPI_SNAP();
 
@@ -333,7 +335,7 @@ int _at_move_job_to_log(job_t *j, bool status, bool process)
 {
 	Datum values[3];	
 	char  nulls[3] = { ' ', ' ', ' ' };	
-	Oid argtypes[4] = { INT4OID, BOOLOID, TEXTOID };
+	Oid argtypes[3] = { INT4OID, BOOLOID, TEXTOID };
 	int ret;
 	const char *sql_process = "WITH moved_rows AS (DELETE from ONLY at_jobs_process a WHERE a.id = $1 RETURNING a.*) INSERT INTO at_jobs_done SELECT *, $2 as status, $3 as reason FROM moved_rows";
 	const char *sql_submitted = "WITH moved_rows AS (DELETE from ONLY at_jobs_submitted a WHERE a.id = $1 RETURNING a.*) INSERT INTO at_jobs_done SELECT *, NULL as start_time, $2 as status, $3 as reason FROM moved_rows";
@@ -356,6 +358,20 @@ int _at_move_job_to_log(job_t *j, bool status, bool process)
 	return ret > 0 ? 1: ret;
 }
 
+int resubmit_at_job(job_t *j, TimestampTz next)
+{
+	Datum values[2];	
+	Oid argtypes[2] = { INT4OID, TIMESTAMPTZOID };
+	int ret;
+	const char *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);
+
+	return ret > 0 ? 1: ret;
+}
+
 int _cron_move_job_to_log(job_t *j, bool status)
 {
 	Datum values[4];	

src/scheduler_job.h

@@ -36,6 +36,8 @@ typedef struct {
 	char *onrollback;
 	char *next_time_statement;
 	bool is_active;
+	int64 attempt;
+	int64 resubmit_limit;
 	char *error;
 } job_t;
 
@@ -52,6 +54,7 @@ job_t *get_at_job(int cron_id, char *nodename, char **perror);
 job_t *get_cron_job(int cron_id, TimestampTz start_at, char *nodename, char **perror);
 int _cron_move_job_to_log(job_t *j, bool status);
 int _at_move_job_to_log(job_t *j, bool status, bool processed);
+int resubmit_at_job(job_t *j, TimestampTz next);
 
 #endif
 

src/scheduler_manager.c

@@ -972,6 +972,12 @@ int scheduler_check_slots(scheduler_manager_ctx_t *ctx, scheduler_manager_pool_t
 				toremove[nremove].reason = shm_data->status == SchdExecutorDone ? RmDone: RmError;
 				nremove++;
 			}
+			else if(shm_data->status == SchdExecutorResubmit)
+			{
+				toremove[nremove].pos = i;
+				toremove[nremove].reason = RmDoneResubmit;
+				nremove++;
+			}
 		}
 	}
 	if(nremove)
@@ -996,6 +1002,11 @@ int scheduler_check_slots(scheduler_manager_ctx_t *ctx, scheduler_manager_pool_t
 					item->wait_worker_to_die = true;
 				}
 			}
+			else if(toremove[i].reason == RmDoneResubmit)
+			{
+				removeJob = true;
+				job_status = true;
+			}
 			else if(toremove[i].reason == RmWaitWorker) /* wait worker to die */
 			{
 				if(GetBackgroundWorkerPid(item->handler, &tmppid) == BGWH_STARTED)
@@ -1066,7 +1077,22 @@ int scheduler_check_slots(scheduler_manager_ctx_t *ctx, scheduler_manager_pool_t
 						}
 					}
 				}
-				move_job_to_log(item->job, job_status, true);
+				if(toremove[i].reason == RmDoneResubmit)
+				{
+					if(item->job->type == AtJob)
+					{
+						resubmit_at_job(item->job, shm_data->next_time);
+					}
+					else
+					{
+						set_job_error(item->job, "cannot resubmit Cron job");
+						move_job_to_log(item->job, false, true);
+					}
+				}
+				else
+				{
+					move_job_to_log(item->job, job_status, true);
+				}
 				STOP_SPI_SNAP();
 
 				last  = p->len - p->free - 1;

src/scheduler_manager.h

@@ -28,6 +28,7 @@ typedef enum {
 	RmTimeout,
 	RmWaitWorker,
 	RmError,
+	RmDoneResubmit,
 	RmDone
 } schd_remove_reason_t;
 

src/scheduler_spi_utils.c

@@ -230,6 +230,17 @@ int get_int_from_spi(int row_n, int pos, int def)
 	return (int)DatumGetInt32(datum);
 }
 
+int64 get_int64_from_spi(int row_n, int pos, int def)
+{
+	Datum datum;
+	bool is_null;
+
+	datum = SPI_getbinval(SPI_tuptable->vals[row_n], SPI_tuptable->tupdesc,
+	                        pos, &is_null);
+	if(is_null) return def;
+	return (int64)DatumGetInt64(datum);
+}
+
 Datum select_onedatumvalue_sql(const char *sql, bool *is_null)
 {
 	int ret;

src/scheduler_spi_utils.h

@@ -18,6 +18,7 @@ void ABORT_SPI_SNAP(void);
 char *_copy_string(char *str);
 TimestampTz get_timestamp_from_spi(int row_n, int pos, TimestampTz def);
 int get_int_from_spi(int row_n, int pos, int def);
+int64 get_int64_from_spi(int row_n, int pos, int def);
 int select_oneintvalue_sql(const char *sql, int d);
 Datum select_onedatumvalue_sql(const char *sql, bool *is_null);
 int select_count_with_args(const char *sql, int n, Oid *argtypes, Datum *values, char *nulls);