Function ProcessBgwTaskFeedback

src/backend/task/pg_cron.c:1657–1757

* ProcessBgwTaskFeedback reads messages from a shared memory queue associated * with the background worker that is executing a given task. If the task is * still running, the function does not block if the queue is empty. Otherwise, * it reads until the end of the queue. */

 * it reads until the end of the queue.
 */
static void
ProcessBgwTaskFeedback(CronTask *task, bool running)
{
	shm_mq_handle *responseq = task->sharedMemoryQueue;
	TimestampTz end_time;

	Size            nbytes;
	void			*data;
	char            msgtype;
	StringInfoData  msg;
	shm_mq_result res;

	end_time = GetCurrentTimestamp();
	/*
	 * Message-parsing routines operate on a null-terminated StringInfo,
	 * so we must construct one.
	 */
	for (;;)
	{
		/* do not wait if the task is running */
		bool nowait = running;

		/* Get next message. */
		res = shm_mq_receive(responseq, &nbytes, &data, nowait);

		if (res != SHM_MQ_SUCCESS)
			break;

		initStringInfo(&msg);
		resetStringInfo(&msg);
		enlargeStringInfo(&msg, nbytes);
		msg.len = nbytes;
		memcpy(msg.data, data, nbytes);
		msg.data[nbytes] = '\0';
		msgtype = pq_getmsgbyte(&msg);
		switch (msgtype)
		{
			case 'N':
			case 'E':
				{
					ErrorData	edata;
					StringInfoData  display_msg;

					pq_parse_errornotice(&msg, &edata);
					initStringInfo(&display_msg);
					bgw_generate_returned_message(&display_msg, edata);

					if (task_log_run)
					{

						if (edata.elevel >= ERROR)
							UpdateJobRunDetail(task->runId, NULL, GetCronStatus(CRON_STATUS_FAILED), display_msg.data, NULL, &end_time);
						else if (running)
							UpdateJobRunDetail(task->runId, NULL, NULL, display_msg.data, NULL, NULL);
						else
							UpdateJobRunDetail(task->runId, NULL, GetCronStatus(CRON_STATUS_SUCCEEDED), display_msg.data, NULL, &end_time);
					}