PANIC in pg_commit_ts slru after crashes

diff --git a/src/backend/access/transam/varsup.c b/src/backend/access/transam/varsup.c
new file mode 100644
index 5efbfbd..f7f8a6d
*** a/src/backend/access/transam/varsup.c
--- b/src/backend/access/transam/varsup.c
***************
*** 33,38 ****
--- 33,41 ----
  /* pointer to "variable cache" in shared memory (set up by shmem.c) */
  VariableCache ShmemVariableCache = NULL;
  
+ int JJ_xid=0;
+ extern int	JJ_vac;
+ 
  
  /*
   * Allocate the next XID for a new transaction or subtransaction.
*************** GetNewTransactionId(bool isSubXact)
*** 168,173 ****
--- 171,181 ----
  	 *
  	 * Extend pg_subtrans and pg_commit_ts too.
  	 */
+ 	{
+ 	int		incr;
+ 	for (incr=0; incr <=JJ_xid; incr++)
+ 	{
+ 	xid = ShmemVariableCache->nextXid;
  	ExtendCLOG(xid);
  	ExtendCommitTs(xid);
  	ExtendSUBTRANS(xid);
*************** GetNewTransactionId(bool isSubXact)
*** 179,184 ****
--- 187,194 ----
  	 * more XIDs until there is CLOG space for them.
  	 */
  	TransactionIdAdvance(ShmemVariableCache->nextXid);
+ 	}
+ 	}
  
  	/*
  	 * We must store the new XID into the shared ProcArray before releasing
*************** SetTransactionIdLimit(TransactionId olde
*** 363,370 ****
  	LWLockRelease(XidGenLock);
  
  	/* Log the info */
! 	ereport(DEBUG1,
! 			(errmsg("transaction ID wrap limit is %u, limited by database with OID %u",
  					xidWrapLimit, oldest_datoid)));
  
  	/*
--- 373,380 ----
  	LWLockRelease(XidGenLock);
  
  	/* Log the info */
! 	ereport(LOG,
! 			(errmsg("JJ transaction ID wrap limit is %u, limited by database with OID %u",
  					xidWrapLimit, oldest_datoid)));
  
  	/*
*************** ForceTransactionIdLimitUpdate(void)
*** 441,446 ****
--- 451,467 ----
  	oldestXidDB = ShmemVariableCache->oldestXidDB;
  	LWLockRelease(XidGenLock);
  
+ 	if (JJ_vac) {
+ 	elog(LOG,"JJ ForceTransactionIdLimitUpdate in %d: !normal %d, !valid %d, follows %d (%u, %u), !exists %d", MyDatabaseId,
+ 		!TransactionIdIsNormal(oldestXid),
+ 		!TransactionIdIsValid(xidVacLimit),
+ 		TransactionIdFollowsOrEquals(nextXid, xidVacLimit),
+ 		nextXid, xidVacLimit,
+  		!SearchSysCacheExists1(DATABASEOID, ObjectIdGetDatum(oldestXidDB))
+ 	);
+ 	};
+ 
+ 
  	if (!TransactionIdIsNormal(oldestXid))
  		return true;			/* shouldn't happen, but just in case */
  	if (!TransactionIdIsValid(xidVacLimit))
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
new file mode 100644
index b99ded5..f3ad6e2
*** a/src/backend/access/transam/xlog.c
--- b/src/backend/access/transam/xlog.c
*************** BootStrapXLOG(void)
*** 5017,5022 ****
--- 5017,5023 ----
  	ShmemVariableCache->oidCount = 0;
  	MultiXactSetNextMXact(checkPoint.nextMulti, checkPoint.nextMultiOffset);
  	AdvanceOldestClogXid(checkPoint.oldestXid);
+ 	//elog(LOG,"JJ SetTransactionIDLimit %d", checkPoint.oldestXid);
  	SetTransactionIdLimit(checkPoint.oldestXid, checkPoint.oldestXidDB);
  	SetMultiXactIdLimit(checkPoint.oldestMulti, checkPoint.oldestMultiDB, true);
  	SetCommitTsLimit(InvalidTransactionId, InvalidTransactionId);
*************** StartupXLOG(void)
*** 6624,6629 ****
--- 6625,6631 ----
  	ShmemVariableCache->oidCount = 0;
  	MultiXactSetNextMXact(checkPoint.nextMulti, checkPoint.nextMultiOffset);
  	AdvanceOldestClogXid(checkPoint.oldestXid);
+ 	//elog(LOG,"JJ SetTransactionIDLimit %d", checkPoint.oldestXid);
  	SetTransactionIdLimit(checkPoint.oldestXid, checkPoint.oldestXidDB);
  	SetMultiXactIdLimit(checkPoint.oldestMulti, checkPoint.oldestMultiDB, true);
  	SetCommitTsLimit(checkPoint.oldestCommitTsXid,
*************** xlog_redo(XLogReaderState *record)
*** 9623,9628 ****
--- 9625,9632 ----
  
  		MultiXactAdvanceOldest(checkPoint.oldestMulti,
  							   checkPoint.oldestMultiDB);
+ 		//elog(LOG,"JJ SetTransactionIDLimit %d", checkPoint.oldestXid);
+ 
  		/*
  		 * No need to set oldestClogXid here as well; it'll be set when we
  		 * redo an xl_clog_truncate if it changed since initialization.
*************** xlog_redo(XLogReaderState *record)
*** 9724,9729 ****
--- 9728,9734 ----
  		 */
  		MultiXactAdvanceOldest(checkPoint.oldestMulti,
  							   checkPoint.oldestMultiDB);
+ 		//elog(LOG,"JJ maybe SetTransactionIDLimit %d", checkPoint.oldestXid);
  		if (TransactionIdPrecedes(ShmemVariableCache->oldestXid,
  								  checkPoint.oldestXid))
  			SetTransactionIdLimit(checkPoint.oldestXid,
diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c
new file mode 100644
index 9fbb0eb..91ec92c
*** a/src/backend/commands/vacuum.c
--- b/src/backend/commands/vacuum.c
*************** int			vacuum_freeze_min_age;
*** 59,64 ****
--- 59,66 ----
  int			vacuum_freeze_table_age;
  int			vacuum_multixact_freeze_min_age;
  int			vacuum_multixact_freeze_table_age;
+ int			JJ_vac=0;
+ 
  
  
  /* A few variables that don't seem worth passing around as parameters */
*************** vacuum_set_xid_limits(Relation rel,
*** 568,573 ****
--- 570,576 ----
  	}
  
  	*freezeLimit = limit;
+ 	if (JJ_vac) elog(LOG,"JJ freezeLimit %d", *freezeLimit);
  
  	/*
  	 * Compute the multixact age for which freezing is urgent.  This is
*************** vacuum_set_xid_limits(Relation rel,
*** 622,627 ****
--- 625,632 ----
  		 * VACUUM schedule, the nightly VACUUM gets a chance to freeze tuples
  		 * before anti-wraparound autovacuum is launched.
  		 */
+ 		if (JJ_vac) elog(LOG,"JJ freeze_min_age %d vacuum_freeze_table_age %d freeze_table_age %d ReadNew %d", freeze_min_age, 
+                            vacuum_freeze_table_age, freeze_table_age,ReadNewTransactionId());
  		freezetable = freeze_table_age;
  		if (freezetable < 0)
  			freezetable = vacuum_freeze_table_age;
*************** vac_update_datfrozenxid(void)
*** 1059,1064 ****
--- 1064,1070 ----
  	 * truncate pg_xact and/or pg_multixact.  Also do it if the shared
  	 * XID-wrap-limit info is stale, since this action will update that too.
  	 */
+ 	if (JJ_vac && dirty) elog(LOG,"JJ updating in %d without call to ForceTransactionIdLimitUpdate", MyDatabaseId);
  	if (dirty || ForceTransactionIdLimitUpdate())
  		vac_truncate_clog(newFrozenXid, newMinMulti,
  						  lastSaneFrozenXid, lastSaneMinMulti);
diff --git a/src/backend/commands/vacuumlazy.c b/src/backend/commands/vacuumlazy.c
new file mode 100644
index b74e493..041cffd
*** a/src/backend/commands/vacuumlazy.c
--- b/src/backend/commands/vacuumlazy.c
***************
*** 64,69 ****
--- 64,70 ----
  #include "utils/tqual.h"
  
  
+ extern int JJ_vac;
  /*
   * Space/time tradeoff parameters: do these need to be user-tunable?
   *
*************** lazy_vacuum_rel(Relation onerel, int opt
*** 243,248 ****
--- 244,251 ----
  	if (options & VACOPT_DISABLE_PAGE_SKIPPING)
  		aggressive = true;
  
+ 	if (JJ_vac) elog(LOG,"JJ aggresive %d, relfrozenid %d", aggressive, onerel->rd_rel->relfrozenxid);
+ 
  	vacrelstats = (LVRelStats *) palloc0(sizeof(LVRelStats));
  
  	vacrelstats->old_rel_pages = onerel->rd_rel->relpages;
diff --git a/src/backend/postmaster/autovacuum.c b/src/backend/postmaster/autovacuum.c
new file mode 100644
index 33ca749..2dd9df2
*** a/src/backend/postmaster/autovacuum.c
--- b/src/backend/postmaster/autovacuum.c
*************** int			autovacuum_vac_cost_delay;
*** 122,127 ****
--- 122,128 ----
  int			autovacuum_vac_cost_limit;
  
  int			Log_autovacuum_min_duration = -1;
+ extern int  JJ_vac;
  
  /* how long to keep pgstat data in the launcher, in milliseconds */
  #define STATS_READ_DELAY 1000
*************** AutoVacWorkerMain(int argc, char *argv[]
*** 1636,1643 ****
  		InitPostgres(NULL, dbid, NULL, InvalidOid, dbname);
  		SetProcessingMode(NormalProcessing);
  		set_ps_display(dbname, false);
- 		ereport(DEBUG1,
- 				(errmsg("autovacuum: processing database \"%s\"", dbname)));
  
  		if (PostAuthDelay)
  			pg_usleep(PostAuthDelay * 1000000L);
--- 1637,1642 ----
*************** AutoVacWorkerMain(int argc, char *argv[]
*** 1645,1651 ****
--- 1644,1654 ----
  		/* And do an appropriate amount of work */
  		recentXid = ReadNewTransactionId();
  		recentMulti = ReadNextMultiXactId();
+ 		if (JJ_vac) ereport(LOG,
+ 				(errmsg("autovacuum: processing database \"%s\" at recent Xid of %u recent mxid of %u", dbname,recentXid,recentMulti)));
  		do_autovacuum();
+ 		if (JJ_vac) ereport(LOG,
+ 				(errmsg("autovacuum: done processing database \"%s\" at recent Xid of %u recent mxid of %u", dbname,ReadNewTransactionId(),ReadNextMultiXactId())));
  	}
  
  	/*
*************** relation_needs_vacanalyze(Oid relid,
*** 2856,2868 ****
  		 * reset, because if that happens, the last vacuum and analyze counts
  		 * will be reset too.
  		 */
- 		elog(DEBUG3, "%s: vac: %.0f (threshold %.0f), anl: %.0f (threshold %.0f)",
- 			 NameStr(classForm->relname),
- 			 vactuples, vacthresh, anltuples, anlthresh);
  
  		/* Determine if this table needs vacuum or analyze. */
  		*dovacuum = force_vacuum || (vactuples > vacthresh);
  		*doanalyze = (anltuples > anlthresh);
  	}
  	else
  	{
--- 2859,2872 ----
  		 * reset, because if that happens, the last vacuum and analyze counts
  		 * will be reset too.
  		 */
  
  		/* Determine if this table needs vacuum or analyze. */
  		*dovacuum = force_vacuum || (vactuples > vacthresh);
  		*doanalyze = (anltuples > anlthresh);
+ 
+ 		if (JJ_vac) elog(LOG, "%s: vac: %.0f (threshold %.0f), anl: %.0f (threshold %.0f) wraparound %d dovaccum %d doanalyze %d",
+ 			 NameStr(classForm->relname),
+ 			 vactuples, vacthresh, anltuples, anlthresh, *wraparound, *dovacuum, *doanalyze);
  	}
  	else
  	{
diff --git a/src/backend/storage/smgr/md.c b/src/backend/storage/smgr/md.c
new file mode 100644
index b0b596d..b903ddb
*** a/src/backend/storage/smgr/md.c
--- b/src/backend/storage/smgr/md.c
***************
*** 68,73 ****
--- 68,75 ----
  #define FILE_POSSIBLY_DELETED(err)	((err) == ENOENT || (err) == EACCES)
  #endif
  
+ int JJ_torn_page=0;
+ 
  /*
   *	The magnetic disk storage manager keeps track of open file
   *	descriptors in its own descriptor pool.  This is done to make it
*************** mdwrite(SMgrRelation reln, ForkNumber fo
*** 805,810 ****
--- 807,813 ----
  	off_t		seekpos;
  	int			nbytes;
  	MdfdVec    *v;
+         static int counter=0;
  
  	/* This assert is too expensive to have on normally ... */
  #ifdef CHECK_WRITE_VS_EXTEND
*************** mdwrite(SMgrRelation reln, ForkNumber fo
*** 830,836 ****
--- 833,850 ----
  				 errmsg("could not seek to block %u in file \"%s\": %m",
  						blocknum, FilePathName(v->mdfd_vfd))));
  
+         if (JJ_torn_page > 0 && counter++ > JJ_torn_page && !RecoveryInProgress()) {
+ 		nbytes = FileWrite(v->mdfd_vfd, buffer, BLCKSZ/3, WAIT_EVENT_DATA_FILE_WRITE);
+ 		ereport(FATAL,
+ 				(errcode(ERRCODE_DISK_FULL),
+ 				 errmsg("could not write block %u of relation %s: wrote only %d of %d bytes",
+ 						blocknum,
+ 						relpath(reln->smgr_rnode, forknum),
+ 						nbytes, BLCKSZ),
+ 				 errhint("JJ is screwing with the database.")));
+         } else {
  	nbytes = FileWrite(v->mdfd_vfd, buffer, BLCKSZ, WAIT_EVENT_DATA_FILE_WRITE);
+ 	}
  
  	TRACE_POSTGRESQL_SMGR_MD_WRITE_DONE(forknum, blocknum,
  										reln->smgr_rnode.node.spcNode,
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
new file mode 100644
index 291bf76..f81e7dd
*** a/src/backend/utils/misc/guc.c
--- b/src/backend/utils/misc/guc.c
***************
*** 111,116 ****
--- 111,119 ----
  /* XXX these should appear in other modules' header files */
  extern bool Log_disconnections;
  extern int	CommitDelay;
+ int	JJ_torn_page;
+ extern int	JJ_xid;
+ extern int	JJ_vac;
  extern int	CommitSiblings;
  extern char *default_tablespace;
  extern char *temp_tablespaces;
*************** static struct config_int ConfigureNamesI
*** 2368,2373 ****
--- 2371,2403 ----
  	},
  
  	{
+ 		{"JJ_torn_page", PGC_USERSET, WAL_SETTINGS,
+ 			gettext_noop("Simulate a torn-page crash after this number of page writes (0 to turn off)"),
+ 			NULL
+ 		},
+ 		&JJ_torn_page,
+ 		0, 0, 100000, NULL, NULL
+ 	},
+ 
+ 	{
+ 		{"JJ_xid", PGC_USERSET, WAL_SETTINGS,
+ 			gettext_noop("Skip this many xid every time we acquire one"),
+ 			NULL
+ 		},
+ 		&JJ_xid,
+ 		0, 0, 1000000, NULL, NULL
+ 	},
+ 
+ 	{
+ 		{"JJ_vac", PGC_USERSET, WAL_SETTINGS,
+ 			gettext_noop("turn on verbose logging"),
+ 			NULL
+ 		},
+ 		&JJ_vac,
+ 		0, 0, 1000000, NULL, NULL
+ 	},
+ 
+ 	{
  		{"commit_siblings", PGC_USERSET, WAL_SETTINGS,
  			gettext_noop("Sets the minimum concurrent open transactions before performing "
  						 "commit_delay."),
diff --git a/src/include/pg_config_manual.h b/src/include/pg_config_manual.h
new file mode 100644
index f3b3529..ef4c90d
*** a/src/include/pg_config_manual.h
--- b/src/include/pg_config_manual.h
***************
*** 300,306 ****
  /*
   * Enable debugging print statements for lock-related operations.
   */
! /* #define LOCK_DEBUG */
  
  /*
   * Enable debugging print statements for WAL-related operations; see
--- 300,306 ----
  /*
   * Enable debugging print statements for lock-related operations.
   */
! #define LOCK_DEBUG 1
  
  /*
   * Enable debugging print statements for WAL-related operations; see

diff --git a/src/backend/access/transam/slru.c b/src/backend/access/transam/slru.c
new file mode 100644
index 7ae7831..87059f4
*** a/src/backend/access/transam/slru.c
--- b/src/backend/access/transam/slru.c
*************** SlruPhysicalReadPage(SlruCtl ctl, int pa
*** 640,645 ****
--- 640,646 ----
  	int			offset = rpageno * BLCKSZ;
  	char		path[MAXPGPATH];
  	int			fd;
+ 	int			sz;
  
  	SlruFileName(ctl, path, segno);
  
*************** SlruPhysicalReadPage(SlruCtl ctl, int pa
*** 677,688 ****
  
  	errno = 0;
  	pgstat_report_wait_start(WAIT_EVENT_SLRU_READ);
! 	if (read(fd, shared->page_buffer[slotno], BLCKSZ) != BLCKSZ)
  	{
  		pgstat_report_wait_end();
  		slru_errcause = SLRU_READ_FAILED;
  		slru_errno = errno;
  		CloseTransientFile(fd);
  		return false;
  	}
  	pgstat_report_wait_end();
--- 678,691 ----
  
  	errno = 0;
  	pgstat_report_wait_start(WAIT_EVENT_SLRU_READ);
! 	sz = read(fd, shared->page_buffer[slotno], BLCKSZ);
! 	if (sz != BLCKSZ)
  	{
  		pgstat_report_wait_end();
  		slru_errcause = SLRU_READ_FAILED;
  		slru_errno = errno;
  		CloseTransientFile(fd);
+ 		elog(LOG,"JJ slru read obtained %d bytes with errno %d", sz, slru_errno);
  		return false;
  	}
  	pgstat_report_wait_end();