fix vanish process

Author: Vladimir Ershov

src/scheduler_job.c

@@ -171,7 +171,8 @@ job_t *get_at_job_for_process(MemoryContext mem, char *nodename, char **error)
 	int i;
 	char *oldpath;
 
-	const char *get_job_sql = "select * from at_jobs_submitted s where ((not exists ( select * from at_jobs_submitted s2 where s2.id = any(s.depends_on)) AND not exists ( select * from 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 and not canceled order by at,  submit_time limit 1 FOR UPDATE SKIP LOCKED";  
+	/* const char *get_job_sql = "select * from at_jobs_submitted s where ((not exists ( select * from at_jobs_submitted s2 where s2.id = any(s.depends_on)) AND not exists ( select * from 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 and not canceled order by at,  submit_time limit 1 FOR UPDATE SKIP LOCKED";  */
+	const char *get_job_sql = "select * from schedule.at_jobs_submitted s where ((not exists ( select * from schedule.at_jobs_submitted s2 where s2.id = any(s.depends_on)) AND not exists ( select * from schedule.at_jobs_process p where p.id = any(s.depends_on)) AND s.depends_on is NOT NULL and s.at IS NULL AND not exists ( select * from schedule.at_jobs_done d where d.id = any(s.depends_on) and d.status=false)) OR ( s.at IS NOT NULL AND  at <= 'now' and (last_start_available is NULL OR last_start_available > 'now'))) and node = $1 and not canceled order by at,  submit_time limit 1 FOR UPDATE SKIP LOCKED ";
 	const char *insert_sql = "insert into at_jobs_process values($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17)";
 	spi_response_t *r;
 	spi_response_t *r2;

src/scheduler_manager.c

@@ -25,6 +25,8 @@
 #include "utils/builtins.h"
 #include "utils/timestamp.h"
 #include <sys/time.h>
+#include "utils/lsyscache.h"
+#include "catalog/namespace.h"
 
 #include "char_array.h"
 #include "sched_manager_poll.h"
@@ -1376,7 +1378,7 @@ int update_cron_texttime(scheduler_manager_ctx_t *ctx, int cron_id, TimestampTz
 	return ret;
 }
 
-int scheduler_vanish_expired_jobs(scheduler_manager_ctx_t *ctx, task_type_t type)
+int scheduler_vanish_expired_jobs(scheduler_manager_ctx_t *ctx, task_type_t type, Oid at_reloid)
 { 
 	job_t *expired;
 	int nexpired  = 0;
@@ -1396,10 +1398,14 @@ int scheduler_vanish_expired_jobs(scheduler_manager_ctx_t *ctx, task_type_t type
 	pgstat_report_activity(STATE_RUNNING, "vanish expired tasks");
 
 	START_SPI_SNAP();
+	if(type == AtJob)
+	{
+		ts_hires = true;
+		scheduler_atjob_id_OID = get_atttype(at_reloid, 1);
+	}
 	expired = type == CronJob ? 
 		get_expired_cron_jobs(ctx->nodename, &nexpired, &is_error):
 		get_expired_at_jobs(ctx->nodename, &nexpired, &is_error);
-	if(type == AtJob) ts_hires = true;
 
 	if(is_error)
 	{
@@ -1760,6 +1766,7 @@ void manager_worker_main(Datum arg)
 	schd_manager_share_t *parent_shared;
 	MemoryContext old = NULL;
 	MemoryContext longTerm;
+	Oid reloid;
 
 
 	CurrentResourceOwner = ResourceOwnerCreate(NULL, "pgpro_scheduler");
@@ -1814,6 +1821,17 @@ void manager_worker_main(Datum arg)
 	longTerm = init_mem_ctx("long term context for slots");
 	ctx = initialize_scheduler_manager_context(longTerm, database, seg);
 
+	START_SPI_SNAP();
+	reloid = RangeVarGetRelid(makeRangeVarFromNameList(
+		stringToQualifiedNameList("schedule.at_jobs_submitted")), NoLock, true);
+	STOP_SPI_SNAP();
+	if(reloid == InvalidOid)
+	{
+		 ereport(ERROR,
+		 	(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+			 errmsg("scheduler manager cannot find jobs table")));
+	}
+
 	start_at_workers(ctx, shared);
 	clean_at_table(ctx);
 	set_slots_stat_report(ctx);
@@ -1850,8 +1868,8 @@ void manager_worker_main(Datum arg)
 				set_slots_stat_report(ctx); 
 				/* if there are any expired jobs to get rid of */
 
-				scheduler_vanish_expired_jobs(ctx, AtJob);
-				scheduler_vanish_expired_jobs(ctx, CronJob);
+				scheduler_vanish_expired_jobs(ctx, AtJob, reloid);
+				scheduler_vanish_expired_jobs(ctx, CronJob, reloid);
 			}
 		}
 

src/scheduler_manager.h

@@ -104,7 +104,7 @@ void fill_cron_array_from_rule(Jsonb *J, const char *name, bit_array_t *ce, int
 bool is_cron_fit_timestamp(bit_array_t *cron, TimestampTz timestamp);
 char **get_dates_array_from_rule(scheduler_task_t *task, int *num);
 int get_integer_from_jsonbval(JsonbValue *ai, int def);
-int scheduler_vanish_expired_jobs(scheduler_manager_ctx_t *ctx, task_type_t type);
+int scheduler_vanish_expired_jobs(scheduler_manager_ctx_t *ctx, task_type_t type, Oid at_reloid);
 int how_many_instances_on_work(scheduler_manager_ctx_t *ctx, job_t *job);
 int insert_at_record(char *nodename, int cron_id, TimestampTz start_at, TimestampTz postpone, char **error);
 int set_job_on_free_slot(scheduler_manager_ctx_t *ctx, job_t *job);

test/perl/runtest.pl

@@ -65,9 +65,9 @@
 	"ALTER SYSTEM SET schedule.enabled = on",
 	"SELECT pg_reload_conf();",
 	"CREATE TABLE test_results( time_mark timestamp, commentary text )",
-	"DROP ROLE IF EXISTS _pgpro_tester",
-	"CREATE ROLE _pgpro_tester",
-	"GRANT INSERT ON test_results TO _pgpro_tester",
+	"DROP ROLE IF EXISTS tester",
+	"CREATE ROLE tester",
+	"GRANT INSERT ON test_results TO tester",
 );
 map { __do_sql($dbh, $_) } @sql2;
 $dbh->disconnect();
@@ -87,7 +87,7 @@
 my $harness = TAP::Harness->new( \%args );
 my @tests = glob( 't/*.t' );
 ### @tests = ('t/jobMaxRunTime.t');
-$harness->runtests(@tests );
+$harness->runtests(@tests);
 
 
 sub __do_sql