Fix reorder buffer memory accounting for toast changes.

While processing toast changes in logical decoding, we rejigger the
tuple change to point to in-memory toast tuples instead to on-disk toast
tuples. And, to make sure the memory accounting is correct, we were
subtracting the old change size and then after re-computing the new tuple,
re-adding its size at the end. Now, if there is any error before we add
the new size, we will release the changes and that will update the
accounting info (subtracting the size from the counters). And we were
underflowing there which leads to an assertion failure in assert enabled
builds and wrong memory accounting in reorder buffer otherwise.

Author: Bertrand Drouvot
Reviewed-by: Amit Kapila
Backpatch-through: 13, where memory accounting was introduced
Discussion: https://postgr.es/m/92b0ee65-b8bd-e42d-c082-4f3f4bf12d34@amazon.com

src/backend/replication/logical/reorderbuffer.c

@@ -290,7 +290,8 @@ static void ReorderBufferToastAppendChunk(ReorderBuffer *rb, ReorderBufferTXN *t
  */
 static Size ReorderBufferChangeSize(ReorderBufferChange *change);
 static void ReorderBufferChangeMemoryUpdate(ReorderBuffer *rb,
-											ReorderBufferChange *change, bool addition);
+											ReorderBufferChange *change,
+											bool addition, Size sz);
 
 /*
  * Allocate a new ReorderBuffer and clean out any old serialized state from
@@ -474,7 +475,8 @@ ReorderBufferReturnChange(ReorderBuffer *rb, ReorderBufferChange *change,
 {
 	/* update memory accounting info */
 	if (upd_mem)
-		ReorderBufferChangeMemoryUpdate(rb, change, false);
+		ReorderBufferChangeMemoryUpdate(rb, change, false,
+										ReorderBufferChangeSize(change));
 
 	/* free contained data */
 	switch (change->action)
@@ -792,7 +794,8 @@ ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn,
 	txn->nentries_mem++;
 
 	/* update memory accounting information */
-	ReorderBufferChangeMemoryUpdate(rb, change, true);
+	ReorderBufferChangeMemoryUpdate(rb, change, true,
+									ReorderBufferChangeSize(change));
 
 	/* process partial change */
 	ReorderBufferProcessPartialChange(rb, txn, change, toast_insert);
@@ -3100,9 +3103,8 @@ ReorderBufferAddNewCommandId(ReorderBuffer *rb, TransactionId xid,
 static void
 ReorderBufferChangeMemoryUpdate(ReorderBuffer *rb,
 								ReorderBufferChange *change,
-								bool addition)
+								bool addition, Size sz)
 {
-	Size		sz;
 	ReorderBufferTXN *txn;
 	ReorderBufferTXN *toptxn;
 
@@ -3127,8 +3129,6 @@ ReorderBufferChangeMemoryUpdate(ReorderBuffer *rb,
 	else
 		toptxn = txn;
 
-	sz = ReorderBufferChangeSize(change);
-
 	if (addition)
 	{
 		txn->size += sz;
@@ -4359,7 +4359,8 @@ ReorderBufferRestoreChange(ReorderBuffer *rb, ReorderBufferTXN *txn,
 	 * update the accounting too (subtracting the size from the counters). And
 	 * we don't want to underflow there.
 	 */
-	ReorderBufferChangeMemoryUpdate(rb, change, true);
+	ReorderBufferChangeMemoryUpdate(rb, change, true,
+									ReorderBufferChangeSize(change));
 }
 
 /*
@@ -4605,17 +4606,23 @@ ReorderBufferToastReplace(ReorderBuffer *rb, ReorderBufferTXN *txn,
 	TupleDesc	toast_desc;
 	MemoryContext oldcontext;
 	ReorderBufferTupleBuf *newtup;
+	Size		old_size;
 
 	/* no toast tuples changed */
 	if (txn->toast_hash == NULL)
 		return;
 
 	/*
-	 * We're going to modify the size of the change, so to make sure the
-	 * accounting is correct we'll make it look like we're removing the change
-	 * now (with the old size), and then re-add it at the end.
+	 * We're going to modify the size of the change. So, to make sure the
+	 * accounting is correct we record the current change size and then after
+	 * re-computing the change we'll subtract the recorded size and then
+	 * re-add the new change size at the end. We don't immediately subtract
+	 * the old size because if there is any error before we add the new size,
+	 * we will release the changes and that will update the accounting info
+	 * (subtracting the size from the counters). And we don't want to
+	 * underflow there.
 	 */
-	ReorderBufferChangeMemoryUpdate(rb, change, false);
+	old_size = ReorderBufferChangeSize(change);
 
 	oldcontext = MemoryContextSwitchTo(rb->context);
 
@@ -4766,8 +4773,11 @@ ReorderBufferToastReplace(ReorderBuffer *rb, ReorderBufferTXN *txn,
 
 	MemoryContextSwitchTo(oldcontext);
 
+	/* subtract the old change size */
+	ReorderBufferChangeMemoryUpdate(rb, change, false, old_size);
 	/* now add the change back, with the correct size */
-	ReorderBufferChangeMemoryUpdate(rb, change, true);
+	ReorderBufferChangeMemoryUpdate(rb, change, true,
+									ReorderBufferChangeSize(change));
 }
 
 /*