fix error messages

set session authorization
fix transaction handling

Author: Vladimir Ershov

src/scheduler_executor.c

@@ -61,8 +61,9 @@ void executor_worker_main(Datum arg)
 	bool success = true;
 	executor_error_t EE;
 	int ret;
-	char *error;
+	char *error = NULL;
 	bool use_pg_vars = true;
+	schd_executor_status_t status;
 	/* int rc = 0; */
 
 	EE.n = 0;
@@ -82,7 +83,7 @@ void executor_worker_main(Datum arg)
 			(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 			 errmsg("executor corrupted dynamic shared memory segment")));
 	}
-	shared->status = SchdExecutorWork;
+	status = shared->status = SchdExecutorWork;
 	SetConfigOption("application_name", "pgp-s executor", PGC_USERSET, PGC_S_SESSION);
 	pgstat_report_activity(STATE_RUNNING, "initialize");
 	init_worker_mem_ctx("ExecutorMemoryContext");
@@ -92,9 +93,28 @@ void executor_worker_main(Datum arg)
 	job = initializeExecutorJob(shared);
 	if(!job)
 	{
-		shared->status = SchdExecutorError;
 		snprintf(shared->message, PGPRO_SCHEDULER_EXECUTOR_MESSAGE_MAX, 
 											"Cannot retrive job information");
+		shared->status = SchdExecutorError;
+		delete_worker_mem_ctx();
+		dsm_detach(seg);
+		proc_exit(0);
+	}
+
+	if(set_session_authorization(job->executor, &error) < 0)
+	{
+		if(error)
+		{
+			snprintf(shared->message, PGPRO_SCHEDULER_EXECUTOR_MESSAGE_MAX,
+				"Cannot set session auth: %s", error);
+			pfree(error);
+		}
+		else
+		{
+			snprintf(shared->message, PGPRO_SCHEDULER_EXECUTOR_MESSAGE_MAX,
+				"Cannot set session auth: unknown error");
+		}
+		shared->status = SchdExecutorError;
 		delete_worker_mem_ctx();
 		dsm_detach(seg);
 		proc_exit(0);
@@ -116,13 +136,15 @@ void executor_worker_main(Datum arg)
 	{
 		pgstat_report_activity(STATE_RUNNING, job->dosql[i]);
 		CHECK_FOR_INTERRUPTS();
-		if(!job->same_transaction) START_SPI_SNAP();
-
+		if(!job->same_transaction)
+		{
+			START_SPI_SNAP();
+		}
 		ret = execute_spi(job->dosql[i], &error);
 		if(ret < 0)
 		{
 			success = false;
-			shared->status = SchdExecutorError;
+			status = SchdExecutorError;
 			if(error)
 			{
 				push_executor_error(&EE, "error on %d: %s", i+1, error);
@@ -139,13 +161,19 @@ void executor_worker_main(Datum arg)
 		}
 		else
 		{
-			if(!job->same_transaction) STOP_SPI_SNAP();
+			if(!job->same_transaction)
+			{
+				STOP_SPI_SNAP();
+			}
 		}
 	}
-	if(shared->status != SchdExecutorError)
+	if(status != SchdExecutorError)
 	{
-		if(job->same_transaction) STOP_SPI_SNAP();
-		shared->status = SchdExecutorDone;
+		if(job->same_transaction)
+		{
+			STOP_SPI_SNAP();
+		}
+		status = SchdExecutorDone;
 	}
 	if(job->next_time_statement)
 	{
@@ -161,12 +189,46 @@ void executor_worker_main(Datum arg)
 	{
 		set_shared_message(shared, &EE);
 	}
+	shared->status = status;
 
 	delete_worker_mem_ctx();
 	dsm_detach(seg);
 	proc_exit(0);
 }
 
+int set_session_authorization(char *username, char **error)
+{
+	Oid types[1] = { TEXTOID };
+	Oid useroid;
+	Datum values[1];
+	bool is_superuser;
+	int ret;
+	char *sql = "select usesysid, usesuper from pg_catalog.pg_user where usename = $1";
+	char buff[1024];
+
+	values[0] = CStringGetTextDatum(username);	
+	START_SPI_SNAP();
+	ret = execute_spi_sql_with_args(sql, 1, types, values, NULL, error);
+
+	if(ret < 0) return ret;
+	if(SPI_processed == 0)
+	{
+		STOP_SPI_SNAP();
+		sprintf(buff, "Cannot find user with name: %s", username);
+		*error = _copy_string(buff);
+
+		return -200;
+	}
+	useroid = get_oid_from_spi(0, 1, 0);
+	is_superuser = get_boolean_from_spi(0, 2, false);
+
+	STOP_SPI_SNAP();
+
+	SetSessionAuthorization(useroid, is_superuser);
+
+	return 1;
+}
+
 void set_shared_message(schd_executor_share_t *shared, executor_error_t *ee)
 {
 	int left = PGPRO_SCHEDULER_EXECUTOR_MESSAGE_MAX;

src/scheduler_executor.h

@@ -17,6 +17,7 @@ typedef enum {
 typedef struct {
 	char database[PGPRO_SCHEDULER_DBNAME_MAX];
 	char nodename[PGPRO_SCHEDULER_NODENAME_MAX];
+	char user[NAMEDATALEN];
 
 	int cron_id;
 	TimestampTz start_at;
@@ -41,6 +42,7 @@ TimestampTz get_next_excution_time(char *sql, executor_error_t *ee);
 int executor_onrollback(job_t *job, executor_error_t *ee);
 void set_pg_var(bool resulti, executor_error_t *ee);
 int push_executor_error(executor_error_t *e, char *fmt, ...)  __attribute__ ((format (gnu_printf, 2, 3)));
+int set_session_authorization(char *username, char **error);
 
 
 #endif

src/scheduler_job.c

@@ -7,6 +7,7 @@
 #include "lib/stringinfo.h"
 #include "scheduler_spi_utils.h"
 #include "utils/timestamp.h"
+#include "utils/builtins.h"
 #include "memutils.h"
 
 job_t *init_scheduler_job(job_t *j)
@@ -22,12 +23,13 @@ job_t *get_jobs_to_do(char *nodename, int *n, int *is_error)
 {
 	job_t *jobs = NULL;
 	int ret, got, i;
-	Oid argtypes[1] = { CSTRINGOID };
+	Oid argtypes[1] = { TEXTOID };
 	Datum values[1];
-	const char *get_job_sql = "select at.start_at, at.last_start_available, at.cron, max_run_time, cron.max_instances from schedule.at at, schedule.cron cron where start_at <= 'now' and not at.active and (last_start_available is NULL OR last_start_available > 'now') and at.cron = cron.id AND cron.node = $1::text";
+	const char *get_job_sql = "select at.start_at, at.last_start_available, at.cron, max_run_time, cron.max_instances, cron.executor from schedule.at at, schedule.cron cron where start_at <= 'now' and not at.active and (last_start_available is NULL OR last_start_available > 'now') and at.cron = cron.id AND cron.node = $1 order by at.start_at";
 
 	*is_error = *n = 0;
-	values[0] = CStringGetDatum(nodename);
+	START_SPI_SNAP();
+	values[0] = CStringGetTextDatum(nodename);
 	ret = SPI_execute_with_args(get_job_sql, 1, argtypes, values, NULL, true, 0);
 	if(ret == SPI_OK_SELECT)
 	{
@@ -45,13 +47,15 @@ job_t *get_jobs_to_do(char *nodename, int *n, int *is_error)
 				jobs[i].timelimit = get_interval_seconds_from_spi(i, 4, 0);
 				jobs[i].max_instances = get_int_from_spi(i, 5, 1);
 				jobs[i].node = _copy_string(nodename);
+				jobs[i].executor = get_text_from_spi(i, 6);
 			}
 		}
 	}
 	else
 	{
 		*is_error = 1;
 	}
+	STOP_SPI_SNAP();
 	return jobs;
 }
 
@@ -107,8 +111,8 @@ job_t *set_job_error(job_t *j, const char *fmt, ...)
 int move_job_to_log(job_t *j, bool status)
 {
 	Datum values[4];	
-	char  nulls[4] = { 0, 0, 0, 0 };	
-	Oid argtypes[4] = { BOOLOID, CSTRINGOID, INT4OID, TIMESTAMPTZOID };
+	char  nulls[4] = { ' ', ' ', ' ', ' ' };	
+	Oid argtypes[4] = { BOOLOID, TEXTOID, INT4OID, TIMESTAMPTZOID };
 	int ret;
 	const char *del_sql = "delete from schedule.at where start_at = $1 and cron = $2";
 	const char *sql = "insert into schedule.log (start_at,  last_start_available, retry, cron, node, started, status, finished, message)  SELECT start_at, last_start_available, retry, cron, node, started, $1 as status, 'now'::timestamp as finished, $2 as message from schedule.at where cron = $3 and start_at = $4";
@@ -118,11 +122,11 @@ int move_job_to_log(job_t *j, bool status)
 	values[0] = BoolGetDatum(status);
 	if(j->error)
 	{
-		values[1] = CStringGetDatum(j->error);
+		values[1] = CStringGetTextDatum(j->error);
 	}
 	else
 	{
-		nulls[1] = 1;
+		nulls[1] = 'n'; 
 	}
 	values[2] = Int32GetDatum(j->cron_id);
 	values[3] = TimestampTzGetDatum(j->start_at);
@@ -134,7 +138,7 @@ int move_job_to_log(job_t *j, bool status)
 		argtypes[1] = INT4OID;
 		values[0] = TimestampTzGetDatum(j->start_at);
 		values[1] = Int32GetDatum(j->cron_id);
-		ret = SPI_execute_with_args(del_sql, 2, argtypes, values, nulls, false, 0);
+		ret = SPI_execute_with_args(del_sql, 2, argtypes, values, NULL, false, 0);
 		if(ret == SPI_OK_DELETE)
 		{
 			return 1;