RE: BUG #19616: pgoutput sends stream abort ('A') to clients that did not enable streaming

Dear Tyler, Andrey,
(Adding Sawada-san in CC)

Good catch and thanks for the report. I confirmed that your reproducer causes an
Assert failure for the debug build. See:

```
(gdb) bt
#0 ReorderBufferMaybeMarkTXNStreamed (rb=0x3385430, txn=0x33a55a8)
at ../postgres/src/backend/replication/logical/reorderbuffer.c:2154
#1 0x0000000000973d05 in ReorderBufferTruncateTXN (rb=0x3385430, txn=0x33a5440,
txn_prepared=false) at ../postgres/src/backend/replication/logical/reorderbuffer.c:1677
#2 0x00000000009740ec in ReorderBufferCheckAndTruncateAbortedTXN (rb=0x3385430, txn=0x33a5440)
at ../postgres/src/backend/replication/logical/reorderbuffer.c:1816
#3 0x0000000000977aa1 in ReorderBufferCheckMemoryLimit (rb=0x3385430)
at ../postgres/src/backend/replication/logical/reorderbuffer.c:3985
#4 0x0000000000972345 in ReorderBufferQueueChange (rb=0x3385430, xid=696, lsn=24970056,
change=0x33a8be0, toast_insert=false)
...
```

Few comments for the code:

```
+ * A top-level transaction is always marked. A subtransaction is marked only
+ * when it has changes and its top-level transaction is already streamed.
```

The last sentence can be "its top-level transaction is already marked as streamed."

```
+ /*
+ * A subtransaction is marked only when it has changes, and only when its
+ * top-level transaction has already been marked as streamed. We never
+ * stream XIDs of empty subxacts, and we must not send an abort for an XID
+ * the downstream has never heard of.
*
- * We do it this way because of aborts - we don't want to send aborts for
- * XIDs the downstream is not aware of. And of course, it always knows
- * about the top-level xact (we send the XID in all messages), but we
- * never stream XIDs of empty subxacts.
+ * The top-level check matters because ReorderBufferTruncateTXN is also
+ * used to discard already-aborted transactions at eviction, where the
+ * top-level xact is not streamed. Marking a subxact there would make a
+ * later abort emit stream_abort to a client that never enabled streaming.
*/
- if (rbtxn_is_toptxn(txn) || (txn->nentries_mem != 0))
+ if (txn->nentries_mem != 0 && rbtxn_is_streamed(rbtxn_get_toptxn(txn)))
```

I feel the code comment might be too detail: second paragraph is not needed
for me.

Best regards,
Hayato Kuroda
FUJITSU LIMITED