text: /*
   * pg_timelord
   *
   * Enable time traveling.
   *
   * TODO:
   *   * work harder on the satisfies function
   *   * handle multiple db for vacuum preventing?
   *   * verify it works
   *
   * This program is open source, licensed under the PostgreSQL license.
   * For license terms, see the LICENSE file.
   *
   */
  #include "postgres.h"
  
  #include <unistd.h>
  
  #include "access/commit_ts.h"
  #include "access/htup_details.h"
  #include "access/xact.h"
  #include "datatype/timestamp.h"
  #include "executor/executor.h"
  #include "executor/spi.h"
  #include "funcapi.h"
  #include "miscadmin.h"
  #include "pgstat.h"
  #include "optimizer/cost.h"
  #include "optimizer/planner.h"
  #include "postmaster/bgworker.h"
  #include "storage/ipc.h"
  #include "storage/lwlock.h"
  #include "storage/procarray.h"
  #include "storage/shmem.h"
  #include "tcop/utility.h"
  #include "utils/builtins.h"
  #include "utils/guc.h"
  #include "utils/ps_status.h"
  #include "utils/snapmgr.h"
  #include "utils/timestamp.h"
  #include "utils/tqual.h"
  
  
  PG_MODULE_MAGIC;
  
  #if PG_VERSION_NUM >= 90300
  #define PGTL_DUMP_FILE		"pg_stat/pg_timelord.stat"
  #else
  #define PGTL_DUMP_FILE		"global/pg_timelord.stat"
  #endif
  
  static const uint32 PGTL_FILE_HEADER = 0x20161025;
  
  static shmem_startup_hook_type prev_shmem_startup_hook = NULL;
  static planner_hook_type prev_planner_hook = NULL;
  static ExecutorStart_hook_type prev_ExecutorStart = NULL;
  static ProcessUtility_hook_type prev_ProcessUtility = NULL;
  
  /*
   * Global shared state
   */
  typedef struct pgtlSharedState
  {
  	LWLockId		lock;			/* protects counter modification */
  	TransactionId	oldestSafeXid;	/* oldest safe transactionid reachable */
  	int8			datname_len;
  	char			datname[NAMEDATALEN];
  } pgtlSharedState;
  
  /* Links to shared memory state */
  static pgtlSharedState *pgtl = NULL;
  
  /* flag set by bgworker signal handler */
  static volatile sig_atomic_t got_sigterm = false;
  
  /*--- Functions --- */
  
  void	_PG_init(void);
  void	_PG_fini(void);
  bool	check_pgtl_ts(char **newval, void **extra, GucSource source);
  void	assign_pgtl_ts(const char *newval, void *extra);
  
  PG_FUNCTION_INFO_V1(pg_timelord_oldest_xact);
  Datum	pg_timelord_oldest_xact(PG_FUNCTION_ARGS);
  
  static void pgtl_main(Datum main_arg);
  static void pgtl_sigterm(SIGNAL_ARGS);
  static void pgtl_shmem_startup(void);
  static void pgtl_shmem_shutdown(int code, Datum arg);
  static PlannedStmt *pgtl_planner(Query *parse,
  								 int cursorOptions,
  								 ParamListInfo boundParams);
  static void pgtl_ExecutorStart(QueryDesc *queryDesc, int eflags);
  static void pgtl_ProcessUtility(Node *parsetree,
  		const char *queryString, ProcessUtilityContext context,
  		ParamListInfo params,
  		DestReceiver *dest, char *completionTag);
  
  static void pgtl_initOldestSafeXid(void);
  static void pgtl_preventWrite(void);
  static void pgtl_restoreWrite(void);
  static bool pgtl_saveShmemState(bool need_lock);
  static void pgtl_setoldestSafeXid(TransactionId newOldest);
  static TransactionId pgtl_getOldestSafeXid(void);
  static void pgtl_setdatname(int8 len, char *datname);
  static bool HeapTupleSatisfiesTimeLord(HeapTuple htup, Snapshot snapshot,
  								Buffer buffer);
  
  static char *pgtl_ts_char;
  static TimestampTz pgtl_ts = 0;
  static char *pgtl_database;
  static int  pgtl_keep_xact;
  static bool old_XactReadOnly;
  static bool old_XactReadOnly_changed = false;
  
  void
  _PG_init(void)
  {
  	BackgroundWorker worker;
  
  	if (!process_shared_preload_libraries_in_progress)
  	{
  		elog(ERROR, "This module can only be loaded via shared_preload_libraries");
  		return;
  	}
  
  	if (!track_commit_timestamp)
  	{
  		elog(ERROR, "I need track_commit_timestamp enabled as a timelord need his tardis!");
  	}
  
  	RequestAddinShmemSpace(MAXALIGN(sizeof(pgtlSharedState)));
  #if PG_VERSION_NUM >= 90600
  	RequestNamedLWLockTranche("pg_timelord", 1);
  #else
  	RequestAddinLWLocks(1);
  #endif
  
  	/* Install hook */
  	prev_shmem_startup_hook = shmem_startup_hook;
  	shmem_startup_hook = pgtl_shmem_startup;
  	prev_planner_hook = planner_hook;
  	planner_hook = pgtl_planner;
  	prev_ExecutorStart = ExecutorStart_hook;
  	ExecutorStart_hook = pgtl_ExecutorStart;
  	prev_ProcessUtility = ProcessUtility_hook;
  	ProcessUtility_hook = pgtl_ProcessUtility;
  
  	DefineCustomStringVariable("pg_timelord.ts",
  							 "Timestamp with time zone destination",
  							 NULL,
  							 &pgtl_ts_char,
  							 "",
  							 PGC_USERSET,
  							 0,
  							 check_pgtl_ts,
  							 assign_pgtl_ts,
  							 NULL);
  
  	DefineCustomStringVariable("pg_timelord.database",
  							   "Defines the database on which prevent VACUUM",
  							   NULL,
  							   &pgtl_database,
  							   "postgres",
  							   PGC_POSTMASTER,
  							   0,
  							   NULL,
  							   NULL,
  							   NULL);
  
  	DefineCustomIntVariable("pg_timelord.keep_xact",
  							"Defines the number of transaction to prevent from cleaning",
  							NULL,
  							&pgtl_keep_xact,
  							200000000,
  							-1,
  							1000000000,
  							PGC_SUSET,
  							0,
  							NULL,
  							NULL,
  							NULL);
  
  	/* Register the worker processes */
  	worker.bgw_flags =
  		BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION;
  	worker.bgw_start_time = BgWorkerStart_RecoveryFinished;		/* Must write to the
  																 * database */
  	worker.bgw_main = pgtl_main;
  	snprintf(worker.bgw_name, BGW_MAXLEN, "pg_timelord");
  	worker.bgw_restart_time = 3;
  	worker.bgw_main_arg = (Datum) 0;
  #if (PG_VERSION_NUM >= 90400)
  	worker.bgw_notify_pid = 0;
  #endif
  	RegisterBackgroundWorker(&worker);
  }
  
  void
  _PG_fini(void)
  {
  	/* uninstall hook */
  	shmem_startup_hook = prev_shmem_startup_hook;
  	planner_hook = prev_planner_hook;
  	ExecutorStart_hook = prev_ExecutorStart;
  	ProcessUtility_hook = prev_ProcessUtility;
  }
  
  Datum
  pg_timelord_oldest_xact(PG_FUNCTION_ARGS)
  {
  	return TransactionIdGetDatum(pgtl_getOldestSafeXid());
  }
  
  static void
  pgtl_main(Datum main_arg)
  {
  	Snapshot		pgtl_snap;
  	char			msg[100];
  
  	/* Establish signal handler before unblocking signals. */
  	pqsignal(SIGTERM, pgtl_sigterm);
  
  	/* We're now ready to receive signals */
  	BackgroundWorkerUnblockSignals();
  
  	/* Connect to postgres database */
  	BackgroundWorkerInitializeConnection(pgtl_database, NULL);
  
  	elog(LOG, "pg_timelord VACUUM preventer© connected on database \"%s\"", pgtl_database);
  
  	set_ps_display("init", false);
  	StartTransactionCommand();
  	SPI_connect();
  
  	/* There doesn't seem to a nice API to set these */
  	XactIsoLevel = XACT_REPEATABLE_READ;
  	XactReadOnly = true;
  
  	pgtl_snap = GetTransactionSnapshot();
  
  	/* detect if new safe xid is needed */
  	pgtl_initOldestSafeXid();
  	pgtl_snap->xmin = pgtl_getOldestSafeXid();
  	Assert(TransactionIdIsNormal(pgtl_getOldestSafeXid()));
  
  	/* start by setting up previous kept xid */
  	PushActiveSnapshot(pgtl_snap);
  	pgstat_report_activity(STATE_IDLEINTRANSACTION, "Setting up pg_timelord...");
  
  	while (!got_sigterm)
  	{
  		/*
  		 * this has to be declared as static (see the hack in GetSnapshotData
  		 * to avoid repeated malloc/free).
  		 */
  		static SnapshotData snap_tmp;
  		TransactionId new_xmin;
  
  		new_xmin = ReadNewTransactionId() - pgtl_keep_xact;
  
  		/* make sure we get a normal xid */
  		if (!TransactionIdIsNormal(new_xmin))
  			new_xmin = FirstNormalTransactionId;
  
  		/* make sure this xid is older than any active xid */
  		/* TODO: make sure we don't get the bgworker xmin... */
  		GetSnapshotData(&snap_tmp);
  		if (TransactionIdPrecedes(snap_tmp.xmin, new_xmin))
  		{
  			new_xmin = snap_tmp.xmin;
  			TransactionIdRetreat(new_xmin);
  		}
  
  		/*
  		 * get a committed xid. This loop could take a long time if your
  		 * application sucks (ie. does a lot of rollback), but you'd deserve it
  		 */
  		while(TransactionIdPrecedes(pgtl_snap->xmin, new_xmin) &&
  			  !TransactionIdDidCommit(new_xmin))
  			TransactionIdRetreat(new_xmin);
  
  		/* don't go further in the past that what we can */
  		if (TransactionIdPrecedes(new_xmin, pgtl_snap->xmin))
  				new_xmin = pgtl_snap->xmin;
  
  		if (pgtl_snap->xmin != new_xmin)
  		{
  			pgtl_snap->xmin = new_xmin;
  			/* these should be unneded, XXX check */
  			//PopActiveSnapshot();
  			//PushActiveSnapshot(pgtl_snap);
  			pgtl_setoldestSafeXid(new_xmin);
  			if (!pgtl_saveShmemState(true))
  				/* XXX should try again or check why it failed? */
  				elog(WARNING, "could not save pg_timelord shmstate");
  
  			if (!ProcArrayInstallImportedXmin(new_xmin, GetTopTransactionId()))
  			{
  				/* this shouldn't happen unless xmin goes forward */
  				elog(PANIC, "error during ProcArrayInstallImportedXmin,"
  						" new_xmin: %d", new_xmin);
  			}
  
  			sprintf(msg, "oldest unvacuumed xid: %u", new_xmin);
  			pgstat_report_activity(STATE_IDLEINTRANSACTION, msg);
  		}
  
  		/* sleep 5 minutes */
  		WaitLatch(MyLatch,
  				WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH,
  				300000
  #if PG_VERSION_NUM >= 100000
  				,PG_WAIT_EXTENSION
  #endif
  		);
  		ResetLatch(MyLatch);
  	}
  
  	proc_exit(1);
  }
  
  static void
  pgtl_sigterm(SIGNAL_ARGS)
  {
  	int save_errno = errno;
  
  	got_sigterm = true;
  	SetLatch(MyLatch);
  
  	errno = save_errno;
  }
  
  static void
  pgtl_shmem_startup(void)
  {
  	bool			found;
  	FILE		   *file;
  	uint32			header;
  	TransactionId	temp_ts;
  	int8			temp_len;
  	char			temp_datname[NAMEDATALEN];
  
  	if (prev_shmem_startup_hook)
  		prev_shmem_startup_hook();
  
  	/* reset in case this is a restart within the postmaster */
  	pgtl = NULL;
  
  	/* Create or attach to the shared memory state */
  	LWLockAcquire(AddinShmemInitLock, LW_EXCLUSIVE);
  
  	/* global access lock */
  	pgtl = ShmemInitStruct("pg_timelord",
  					sizeof(pgtlSharedState),
  					&found);
  
  	if (!found)
  	{
  		/* First time through ... */
  #if PG_VERSION_NUM >= 90600
  		pgtl->lock = &(GetNamedLWLockTranche("pg_timelord"))->lock;
  #else
  		pgtl->lock = LWLockAssign();
  #endif
  	}
  
  	LWLockRelease(AddinShmemInitLock);
  
  	if (!IsUnderPostmaster)
  		on_shmem_exit(pgtl_shmem_shutdown, (Datum) 0);
  
  	/*
  	 * Done if some other process already completed our initialization.
  	 */
  	if (found)
  		return;
  
  	/* Load stat file, don't care about locking */
  	file = AllocateFile(PGTL_DUMP_FILE, PG_BINARY_R);
  	if (file == NULL)
  	{
  		if (errno == ENOENT)
  		{
  			/*
  			 * no stat file, probably initialization. A safe xid will be set
  			 * at first query execution, save configured database as reference
  			 * database.
  			 */
  			pgtl_setoldestSafeXid(BootstrapTransactionId);
  			pgtl_setdatname(strlen(pgtl_database), pgtl_database);
  			return;			/* ignore not-found error */
  		}
  		goto error;
  	}
  
  	/* check is header is valid */
  	if (fread(&header, sizeof(uint32), 1, file) != 1 ||
  		header != PGTL_FILE_HEADER)
  		goto error;
  
  	/* read saved oldestSafeXid and datname */
  	if ((fread(&temp_ts, sizeof(TransactionId), 1, file) != 1) ||
  		(fread(&temp_len, sizeof(int8), 1, file) != 1) ||
  		(fread(&temp_datname, 1, temp_len + 1, file) != temp_len + 1))
  		goto error;
  
  	Assert(temp_len <= NAMEDATALEN);
  
  	/* detect if database changed */
  	if (strcmp(pgtl_database, temp_datname) != 0)
  	{
  		/* database changed, force new safe xid */
  		pgtl_setoldestSafeXid(BootstrapTransactionId);
  	}
  	else
  	{
  		/* restore the oldest safe transaction id */
  		pgtl_setoldestSafeXid(temp_ts);
  	}
  	/* save reference database in shmem */
  	pgtl_setdatname(temp_len, temp_datname);
  
  	FreeFile(file);
  
  	/*
  	 * Don't remove the file so it's included in backups/replication slaves,
  	 * etc. This file will be rewritten as needed by the background worker or
  	 * during shutdown
  	 */
  
  	return;
  
  error:
  	ereport(LOG,
  			(errcode_for_file_access(),
  			 errmsg("could not read pg_timelord file \"%s\": %m",
  					PGTL_DUMP_FILE)));
  	if (file)
  		FreeFile(file);
  	/* delete bogus file, don't care of errors in this case */
  	unlink(PGTL_DUMP_FILE);
  }
  
  static void
  pgtl_shmem_shutdown(int code, Datum arg)
  {
  	/* Don't try to dump during a crash. */
  	if (code)
  		return;
  
  	if (!pgtl)
  		return;
  
  	/* no lock here */
  	pgtl_saveShmemState(false);
  }
  
  bool
  check_pgtl_ts(char **newval, void **extra, GucSource source)
  {
  	TimestampTz newTs = 0;
  	TransactionId oldest;
  	TimestampTz oldestTs = 0;
  
  	if (!track_commit_timestamp)
  	{
  		elog(ERROR, "I need track_commit_timestamp enabled as a timelord need his tardis!");
  		return false;
  	}
  
  	if (!pgtl)
  	{
  		/* don't allow non empty default */
  		if (strcmp(*newval, "") != 0)
  			return false;
  	}
  	else
  	{
  		/* shmem avail and asked for ts, let's check a safe xid exists first */
  
  		oldest = pgtl_getOldestSafeXid();
  
  		if (!TransactionIdIsNormal(oldest))
  		{
  			elog(ERROR, "No safe point is the past available yet");
  			return false;
  		}
  	}
  
  	if (IsTransactionBlock())
  	{
  		/*
  		 * no time travel if in a transaction
  		 * XXX allow if transaction began while time traveling and changing
  		 * date?
  		 */
  		elog(ERROR, "No time travel inside a transaction");
  		return false;
  	}
  
  	if (strcmp(*newval, "") != 0)
  	{
  		/* new char is needed, don't know why *newval doesn't work */
  		char *str = pstrdup(*newval);
  
  		newTs = DatumGetTimestampTz(DirectFunctionCall3(timestamptz_in,
  					CStringGetDatum(str),
  					ObjectIdGetDatum(InvalidOid),
  					Int32GetDatum(-1)));
  		pfree(str);
  
  		/* check if ts is recent enough */
  		if (!TransactionIdGetCommitTsData(oldest, &oldestTs, NULL))
  		{
  			if (TransactionIdEquals(oldest, ReadNewTransactionId()))
  				elog(ERROR, "There is no previous commit stored");
  			else
  				elog(ERROR, "Could not check data availability for ts %s", *newval);
  			return false;
  		}
  
  		if (timestamp_cmp_internal(newTs, oldestTs) < 0)
  		{
  			ereport(ERROR,
  					(errmsg("Requested ts %s is too old", *newval),
  					errdetail("Oldest ts available: %s",
  							  timestamptz_to_str(oldestTs))));
  			return false;
  		}
  	}
  
  	*extra = malloc(sizeof(TimestampTz));
  	if (!*extra)
  		return false;
  	*((TimestampTz *) *extra) = newTs;
  
  	return true;
  }
  
  void
  assign_pgtl_ts(const char *newval, void *extra)
  {
  	pgtl_ts = *((TimestampTz *) extra);
  }
  
  static PlannedStmt *pgtl_planner(Query *parse,
  								 int cursorOptions,
  								 ParamListInfo boundParams)
  {
  	PlannedStmt *result;
  	bool prev_enable_bitmapscan = enable_bitmapscan;
  	bool prev_enable_indexonlyscan = enable_indexonlyscan;
  
  	if (pgtl_ts != 0)
  	{
  		/*
  		 * bitmap scan does not support non-core MVCC snapshot, see Assert in
  		 * ExecInitBitmapHeapScan() and comment at top of nodeBitmapHeapscan.c
  		 */
  		enable_bitmapscan = false;
  		/* index-only scan does not support non-core MVCC snapshot, see comment
  		 * above the elog(ERROR) in IndexOnlyNext()
  		 */
  		enable_indexonlyscan = false;
  	}
  
  	if (prev_planner_hook)
  		result = (*prev_planner_hook) (parse, cursorOptions, boundParams);
  	else
  		result = standard_planner(parse, cursorOptions, boundParams);
  
  	if (pgtl_ts != 0)
  	{
  		/* restore planner parameters */
  		enable_bitmapscan = prev_enable_bitmapscan;
  		enable_indexonlyscan = prev_enable_indexonlyscan;
  	}
  
  	return result;
  }
  
  static void
  pgtl_ExecutorStart (QueryDesc *queryDesc, int eflags)
  {
  	pgtl_preventWrite();
  
  	/* do not mess with non MVCC snapshots, like index build :) */
  	if (pgtl_ts != 0 && IsMVCCSnapshot(queryDesc->snapshot))
  		queryDesc->snapshot->satisfies = HeapTupleSatisfiesTimeLord;
  
  	if (prev_ExecutorStart)
  		prev_ExecutorStart(queryDesc, eflags);
  	else
  		standard_ExecutorStart(queryDesc, eflags);
  
  	pgtl_restoreWrite();
  }
  
  static void
  pgtl_ProcessUtility(Node *parsetree,
  		const char *queryString, ProcessUtilityContext context,
  		ParamListInfo params,
  		DestReceiver *dest, char *completionTag)
  {
  	pgtl_preventWrite();
  
  	/* forbid most of utility command in the past! */
  	if (pgtl_ts != 0)
  	{
  		switch (nodeTag(parsetree))
  		{
  			case T_TransactionStmt:
  				{
  					TransactionStmt *stmt = (TransactionStmt *) parsetree;
  
  					/* error out if explicitely asked rw transaction */
  					if (stmt->kind == TRANS_STMT_BEGIN ||
  						stmt->kind == TRANS_STMT_START)
  					{
  						ListCell *lc;
  
  						foreach(lc, stmt->options)
  						{
  							DefElem    *item = (DefElem *) lfirst(lc);
  
  							if (strcmp(item->defname, "transaction_read_only") == 0)
  							{
  								A_Const *con = (A_Const *) item->arg;
  
  								Assert(IsA(item->arg, A_Const) && IsA(&con->val, Integer));
  								if (!intVal(&con->val))
  								{
  									elog(ERROR, "You cannot ask for read-write transaction in the past!");
  								}
  							}
  						}
  					}
  					/*
  					 * I keep the code above just in case I want to allow it
  					 * again sometime in the future, but for now just forbid
  					 * any transaction in the past
  					 */
  					elog(ERROR, "You cannot use TCL in the past!");
  				}
  			case T_VariableSetStmt:
  			case T_VariableShowStmt:
  			case T_ExplainStmt:
  			case T_NotifyStmt:
  			case T_ListenStmt:
  			case T_UnlistenStmt:
  			case T_CopyStmt:
  			case T_PrepareStmt:
  			case T_DeallocateStmt:
  			case T_PlannedStmt:
  			case T_ClosePortalStmt:
  			case T_FetchStmt:
  			case T_DoStmt:
  				/* whitelist of utility command */
  				break;
  			default:
  				elog(ERROR, "You cannot run utility command \"%s\" in the past!",
  						CreateCommandTag(parsetree));
  		}
  	}
  
  	if (IsA(parsetree, VacuumStmt))
  	{
  		VacuumStmt *vacstmt = (VacuumStmt *) parsetree;
  
  		if (vacstmt->options & VACOPT_VERBOSE)
  			elog(NOTICE, "pg_timelord is blocking VACUUM for xid >= %u",
  					pgtl_getOldestSafeXid());
  	}
  
  	if (prev_ProcessUtility)
  	(*prev_ProcessUtility) (parsetree, queryString,
  								context, params,
  								dest, completionTag);
  	else
  		standard_ProcessUtility(parsetree, queryString,
  								context, params,
  								dest, completionTag);
  
  	pgtl_restoreWrite();
  }
  
  static void
  pgtl_initOldestSafeXid(void)
  {
  	/* test without lock first */
  	if (TransactionIdEquals(pgtl->oldestSafeXid, BootstrapTransactionId))
  	{
  		/* new test with the lock, and set value if needed */
  		LWLockAcquire(pgtl->lock, LW_EXCLUSIVE);
  		if (pgtl->oldestSafeXid == BootstrapTransactionId)
  		{
  			pgtl->oldestSafeXid = ReadNewTransactionId();
  			elog(WARNING, "new safe xid set: %u", pgtl->oldestSafeXid);
  		}
  		LWLockRelease(pgtl->lock);
  	}
  }
  
  static void
  pgtl_preventWrite(void)
  {
  	if (pgtl_ts != 0)
  	{
  		old_XactReadOnly = XactReadOnly;
  		XactReadOnly = true;
  		old_XactReadOnly_changed = true;
  	}
  	else
  		/* bgworker should have done it already */
  		pgtl_initOldestSafeXid();
  }
  
  static void
  pgtl_restoreWrite(void)
  {
  	if (old_XactReadOnly_changed)
  	{
  		XactReadOnly = old_XactReadOnly;
  		old_XactReadOnly_changed = false;
  	}
  }
  
  /* callse must take care of taking the lock */
  static bool
  pgtl_saveShmemState(bool need_lock)
  {
  	FILE   *file;
  	bool	lock_taken = false;
  
  	if (!pgtl)
  		return false;
  
  	if (need_lock)
  	{
  		LWLockAcquire(pgtl->lock, LW_SHARED);
  		lock_taken = true;
  	}
  
  	file = AllocateFile(PGTL_DUMP_FILE ".tmp", PG_BINARY_W);
  	if (file == NULL)
  		goto error;
  
  	if ((fwrite(&PGTL_FILE_HEADER, sizeof(uint32), 1, file) != 1) ||
  		(fwrite(&pgtl->oldestSafeXid, sizeof(TransactionId), 1, file) != 1) ||
  		(fwrite(&pgtl->datname_len, sizeof(int8), 1, file) != 1) ||
  		(fwrite(&pgtl->datname, 1, pgtl->datname_len + 1, file) != pgtl->datname_len + 1))
  		goto error;
  
  	if (FreeFile(file))
  	{
  		file = NULL;
  		goto error;
  	}
  
  	/*
  	 * Remove old file and rename file inplace
  	 */
  	unlink(PGTL_DUMP_FILE);
  	/* if crash happens here, you're really not lucky */
  	if (rename(PGTL_DUMP_FILE ".tmp", PGTL_DUMP_FILE) != 0)
  	{
  		ereport(LOG,
  				(errcode_for_file_access(),
  				 errmsg("could not rename pg_timelord file \"%s\": %m",
  						PGTL_DUMP_FILE ".tmp")));
  
  		goto error;
  	}
  
  	if (need_lock && lock_taken)
  		LWLockRelease(pgtl->lock);
  
  	return true;
  
  error:
  	ereport(LOG,
  			(errcode_for_file_access(),
  			 errmsg("could not write pg_timelord file \"%s\": %m",
  					PGTL_DUMP_FILE)));
  
  	if (file)
  		FreeFile(file);
  	unlink(PGTL_DUMP_FILE);
  
  	if (lock_taken)
  		LWLockRelease(pgtl->lock);
  	return false;
  }
  
  static void
  pgtl_setoldestSafeXid(TransactionId newOldest)
  {
  	Assert(TransactionIdIsValid(newOldest));
  	Assert(TransactionIdPrecedesOrEquals(pgtl->oldestSafeXid, newOldest));
  
  	if (TransactionIdEquals(newOldest, BootstrapTransactionId))
  		elog(WARNING, "Forcing new safe xid at next transaction");
  
  	LWLockAcquire(pgtl->lock, LW_EXCLUSIVE);
  	pgtl->oldestSafeXid = newOldest;
  	LWLockRelease(pgtl->lock);
  }
  
  static TransactionId
  pgtl_getOldestSafeXid(void)
  {
  	TransactionId xmin;
  
  	LWLockAcquire(pgtl->lock, LW_SHARED);
  	xmin = pgtl->oldestSafeXid;
  	LWLockRelease(pgtl->lock);
  
  	return xmin;
  }
  
  static void
  pgtl_setdatname(int8 len, char *datname)
  {
  	LWLockAcquire(pgtl->lock, LW_EXCLUSIVE);
  	pgtl->datname_len = len;
  	if (len > 0)
  		strncpy(pgtl->datname, datname, len + 1);
  	LWLockRelease(pgtl->lock);
  }
  
  static bool
  HeapTupleSatisfiesTimeLord(HeapTuple htup, Snapshot snapshot,
  					   Buffer buffer)
  {
  	/* naive way FTW */
  	HeapTupleHeader tuple = htup->t_data;
  	TransactionId xmin = HeapTupleHeaderGetXmin(tuple);
  	TransactionId xmax = HeapTupleHeaderGetRawXmax(tuple);
  
  	Assert(TransactionIdIsValid(xmin));
  
  	/* frozen and bootstrap xid are always visible */
  	if (TransactionIdIsNormal(xmin) && (!HeapTupleHeaderXminFrozen(tuple)))
  	{
  		/* really inserted ? */
  		if (HeapTupleHeaderXminCommitted(tuple) && TransactionIdDidCommit(xmin))
  		{
  			TimestampTz insert_ts;
  
  			if (!TransactionIdGetCommitTsData(xmin, &insert_ts, NULL))
  				/* too old, too bad */
  				elog(ERROR, "You requested too old snapshot");
  
  			if (timestamp_cmp_internal(insert_ts, pgtl_ts) > 0)
  				/* line created after requested ts */
  				return false;
  		}
  		else /* line was never inserted */
  			return false;
  	}
  
  	/* at this point, xmin has committed and is visible, what about xmax ? */
  
  	/*
  	 * quick exit if no xmax, probably not needed since HEAP_XMAX_COMMITTED
  	 * should not be true
  	 */
  	if (!TransactionIdIsValid(xmax))
  			return true;
  
  	/* check hint bit first */
  	if ((tuple->t_infomask & HEAP_XMAX_COMMITTED) ||
  		/* check clog */
  		TransactionIdDidCommit(xmax))
  	{
  		TimestampTz delete_ts;
  
  		/* I think this test is stupid */
  		if (xmax != FrozenTransactionId)
  		{
  			if (!TransactionIdGetCommitTsData(xmax, &delete_ts, NULL))
  				/* too old, too bad */
  				elog(ERROR, "You requested too old snapshot");
  
  			if (timestamp_cmp_internal(delete_ts, pgtl_ts) > 0)
  				/* line deleted after requested ts */
  				return true;
  			else
  				return false;
  		}
  	}
  
  	return true;
  }
  