From 346713e34944002b058681799cd1ebbaff9e1b08 Mon Sep 17 00:00:00 2001
From: Georges Berenger <gberenger@meta.com>
Date: Thu, 28 Dec 2023 00:44:47 -0800
Subject: [PATCH] Be more conservative when collecting old records

Summary: Because records of different streams may have the same timestamp, when collecting "old" records to write, it's safer to exclude the limit time, so that if a record is later produced with the same timestamp as the limit given, it is more likely to be sorted correctly when written out to disk. This condition can happen when copying files with vrstool.

Reviewed By: kiminoue7

Differential Revision: D52425279

fbshipit-source-id: 166fe9049c5913a238837b5bbebeb598326677b3
---
 vrs/RecordManager.cpp          |  2 +-
 vrs/test/RecordManagerTest.cpp |  8 +++++---
 vrs/test/RecordTest.cpp        | 14 +++++++-------
 3 files changed, 13 insertions(+), 11 deletions(-)

diff --git a/vrs/RecordManager.cpp b/vrs/RecordManager.cpp
index 3ba7eca1..72d64ff7 100644
--- a/vrs/RecordManager.cpp
+++ b/vrs/RecordManager.cpp
@@ -172,7 +172,7 @@ void RecordManager::collectOldRecords(double maxAge, list<Record*>& outCollected
   if (!activeRecords_.empty()) {
     auto iterator = upper_bound(
         activeRecords_.begin(), activeRecords_.end(), maxAge, [](double age, Record* record) {
-          return age < record->timestamp_;
+          return record->timestamp_ >= age;
         });
 
     // Move a range without copying elements.
diff --git a/vrs/test/RecordManagerTest.cpp b/vrs/test/RecordManagerTest.cpp
index 4196eaf5..124df357 100644
--- a/vrs/test/RecordManagerTest.cpp
+++ b/vrs/test/RecordManagerTest.cpp
@@ -41,9 +41,9 @@ TEST(RecordManager, CollectOldRecords) {
   // Verify that we can pull a subsection of the records.
   list<Record*> recordData;
   manager.collectOldRecords(1.33, recordData);
-  EXPECT_EQ(recordData.size(), 134);
+  EXPECT_EQ(recordData.size(), 133);
   EXPECT_DOUBLE_EQ(recordData.front()->getTimestamp(), 0.0);
-  EXPECT_DOUBLE_EQ(recordData.back()->getTimestamp(), 1.33);
+  EXPECT_LT(recordData.back()->getTimestamp(), 1.33);
   for (auto* r : recordData) {
     r->recycle();
   }
@@ -52,9 +52,11 @@ TEST(RecordManager, CollectOldRecords) {
   list<Record*> noRecords;
   manager.collectOldRecords(0.74, noRecords);
   EXPECT_TRUE(noRecords.empty());
+  manager.collectOldRecords(1.33, noRecords);
+  EXPECT_TRUE(noRecords.empty());
 
   // Purge the rest of the records.
-  EXPECT_EQ(1000 - 134, manager.purgeOldRecords(10.0));
+  EXPECT_EQ(1000 - recordData.size(), manager.purgeOldRecords(10.0));
 
   // Call it again. There shouldn't be anymore.
   manager.collectOldRecords(1000.0, noRecords);
diff --git a/vrs/test/RecordTest.cpp b/vrs/test/RecordTest.cpp
index 61476350..e8f6f83f 100644
--- a/vrs/test/RecordTest.cpp
+++ b/vrs/test/RecordTest.cpp
@@ -45,7 +45,7 @@ size_t collect(
     list<Record*>& records = batch.back().second;
     r.first->collectOldRecords(maxTime, records);
     for (auto record : records) {
-      EXPECT_LE(record->getTimestamp(), maxTime);
+      EXPECT_LT(record->getTimestamp(), maxTime);
     }
     count += records.size();
   }
@@ -150,9 +150,9 @@ TEST_F(RecordTester, addRecordBatchesToSortedRecordsTester) {
   vector<pair<RecordManager*, StreamId>> recordManagerBOnly{{&recordManagerB, idB}};
   vector<pair<RecordManager*, StreamId>> recordManagerCOnly{{&recordManagerC, idC}};
 
-  EXPECT_EQ(collect(batches, recordManagersAll, 5), 86);
+  EXPECT_EQ(collect(batches, recordManagersAll, 5), 85);
   RecordFileWriterTester::addRecordBatchesToSortedRecords(batches, sr);
-  EXPECT_EQ(sr.size(), 86);
+  EXPECT_EQ(sr.size(), 85);
   EXPECT_TRUE(isProperlySorted(sr));
   batches.clear();
 
@@ -160,22 +160,22 @@ TEST_F(RecordTester, addRecordBatchesToSortedRecordsTester) {
   EXPECT_EQ(collect(batches, recordManagerCOnly, 8), 20);
   recordManagerA.createRecord(6.25, Record::Type::DATA, 1, DataSource());
   recordManagerB.createRecord(4, Record::Type::DATA, 1, DataSource());
-  EXPECT_EQ(collect(batches, recordManagersAll, 10), 38);
+  EXPECT_EQ(collect(batches, recordManagersAll, 10), 37);
   RecordFileWriterTester::addRecordBatchesToSortedRecords(batches, sr);
-  EXPECT_EQ(sr.size(), 177);
+  EXPECT_EQ(sr.size(), 175);
   EXPECT_TRUE(isProperlySorted(sr));
   batches.clear();
 
   // don't collect anything this time
   EXPECT_EQ(collect(batches, recordManagersAll, 10), 0);
   RecordFileWriterTester::addRecordBatchesToSortedRecords(batches, sr);
-  EXPECT_EQ(sr.size(), 177);
+  EXPECT_EQ(sr.size(), 175);
   EXPECT_TRUE(isProperlySorted(sr));
   batches.clear();
 
   recordManagerA.createRecord(2.5, Record::Type::DATA, 1, DataSource());
   recordManagerA.createRecord(3.5, Record::Type::DATA, 1, DataSource());
-  EXPECT_EQ(collect(batches, recordManagersAll, 100), 277);
+  EXPECT_EQ(collect(batches, recordManagersAll, 100), 279);
   RecordFileWriterTester::addRecordBatchesToSortedRecords(batches, sr);
   EXPECT_EQ(sr.size(), 454);
   EXPECT_TRUE(isProperlySorted(sr));