fix: filter per-thread archives in RAC online mode (#14)

archGetLogOnline() queried V$ARCHIVED_LOG from minSeq across all threads
but created Parser objects for every row, including already-processed
archives from the ahead thread. On a 2-node RAC with ~1600 seq divergence
this allocated ~1600 unnecessary 1MB Parser chunks, causing OOM.

Add per-thread filtering on query results, matching archGetLogPath() and
archGetLogList() logic. Verified: RSS dropped from 1012MB to 115MB.

Author: rophy

UPSTREAM-CHANGES.md

@@ -37,7 +37,7 @@ Multi-thread redo log processing for Oracle RAC.
 | 1a2d316b | LOB INSERT preserved from phantom undo in streaming | Transaction.cpp | [#10](https://github.com/rophy/olr/issues/10) |
 | ddf6bc04 | skip truncated URP null bitmap instead of abort | OpCode0504.cpp | — |
 | 7cbd0580 | decode ROWID column type (type# 69) | Builder.cpp, SysCol.h, BuilderJson.* | [#15](https://github.com/rophy/olr/issues/15) |
-| — | OOM null-deref crash: throw instead of returning nullptr | Ctx.cpp, MemoryManager.cpp | [#14](https://github.com/rophy/olr/issues/14) |
+| — | RAC archive query creates parsers for already-processed logs | ReplicatorOnline.cpp | [#14](https://github.com/rophy/olr/issues/14) |
 
 **Upstream PR candidates:** All — these are correctness fixes
 

src/replicator/ReplicatorOnline.cpp

@@ -1601,7 +1601,9 @@ namespace OpenLogReplicator {
         if (!replicatorOnline->checkConnection())
             return;
 
-        // Find minimum sequence across all threads for the SQL filter
+        // Use minimum sequence across all threads for the SQL filter so the
+        // query finds archives for the lagging thread.  Per-thread filtering
+        // below prevents creating Parsers for already-processed archives.
         Seq minSeq = Seq::none();
         {
             std::unique_lock const lck(replicator->metadata->mtxCheckpoint);
@@ -1610,7 +1612,6 @@ namespace OpenLogReplicator {
                     minSeq = state.sequence;
             }
         }
-        // Fallback to legacy sequence if threadStates is empty
         if (minSeq == Seq::none())
             minSeq = replicator->metadata->sequence;
 
@@ -1639,6 +1640,16 @@ namespace OpenLogReplicator {
 
             int ret = stmt.executeQuery();
             while (ret != 0) {
+                const auto threadNum = static_cast<uint16_t>(thread);
+
+                // Filter per-thread: skip archives already processed by this thread.
+                // Matches archGetLogPath()/archGetLogList() logic (Replicator.cpp:531-533).
+                const Seq threadSeq = replicator->metadata->getSequence(threadNum);
+                if (threadSeq != Seq::none() && sequence < threadSeq) {
+                    ret = stmt.next();
+                    continue;
+                }
+
                 std::string mappedPath(path.data());
                 replicator->applyMapping(mappedPath);
 
@@ -1647,8 +1658,8 @@ namespace OpenLogReplicator {
                 parser->firstScn = firstScn;
                 parser->nextScn = nextScn;
                 parser->sequence = sequence;
-                parser->thread = static_cast<uint16_t>(thread);
-                replicator->archiveRedoQueues[static_cast<uint16_t>(thread)].push(parser);
+                parser->thread = threadNum;
+                replicator->archiveRedoQueues[threadNum].push(parser);
                 ret = stmt.next();
             }
         }