From 1a9440ec3c1669c5e49825dde70b9cbede3f0de0 Mon Sep 17 00:00:00 2001
From: Dan Goodliffe <dan.goodliffe@octal.co.uk>
Date: Thu, 9 Oct 2025 15:48:28 +0100
Subject: Fix premature remembering of saved entity ids

Don't persist entity ids saved to the DB until the transaction is
committed. Prevents the issue where a later DB operation fails, the
transaction is rolled back, but we still think the entity has been
saved.
---
 src/ingestor.cpp | 32 ++++++++++++++++++++++----------
 src/ingestor.hpp |  5 +++--
 2 files changed, 25 insertions(+), 12 deletions(-)

(limited to 'src')

diff --git a/src/ingestor.cpp b/src/ingestor.cpp
index f965d1d..4cd7859 100644
--- a/src/ingestor.cpp
+++ b/src/ingestor.cpp
@@ -164,20 +164,30 @@ namespace WebStat {
 	void
 	Ingestor::ingestLogLine(DB::Connection * dbconn, const std::string_view line)
 	{
+		auto rememberNewEntityIds = [this](const auto & ids) {
+			existingEntities.insert_range(ids | std::views::take_while(&std::optional<Crc32Value>::has_value)
+					| std::views::transform([](auto && value) {
+						  return *value;
+					  }));
+		};
 		if (auto result = scanLogLine(line)) {
 			linesParsed++;
 			const auto values = crc32ScanValues(result->values());
-			std::optional<DB::TransactionScope> dbtx;
-			if (const auto newEnts = newEntities(values); newEnts.front()) {
-				dbtx.emplace(*dbconn);
-				storeEntities(dbconn, newEnts);
+			NewEntityIds ids;
+			{
+				std::optional<DB::TransactionScope> dbtx;
+				if (const auto newEnts = newEntities(values); newEnts.front()) {
+					dbtx.emplace(*dbconn);
+					ids = storeEntities(dbconn, newEnts);
+				}
+				storeLogLine(dbconn, values);
 			}
-			storeLogLine(dbconn, values);
+			rememberNewEntityIds(ids);
 		}
 		else {
 			linesDiscarded++;
 			const auto unparsableLine = toEntity(line, EntityType::UnparsableLine);
-			storeEntities(dbconn, {unparsableLine});
+			rememberNewEntityIds(storeEntities(dbconn, {unparsableLine}));
 		}
 	}
 
@@ -270,15 +280,16 @@ namespace WebStat {
 		return rtn;
 	}
 
-	void
+	Ingestor::NewEntityIds
 	Ingestor::storeEntities(DB::Connection * dbconn, const std::span<const std::optional<Entity>> values) const
 	{
 		static constexpr std::array ENTITY_TYPE_VALUES {
 				"host", "virtual_host", "path", "query_string", "referrer", "user_agent", "unparsable_line"};
 
 		auto insert = dbconn->modify(SQL::ENTITY_INSERT, SQL::ENTITY_INSERT_OPTS);
-		std::ranges::for_each(
-				values | std::views::take_while(&std::optional<Entity>::has_value), [this, &insert](auto && entity) {
+		NewEntityIds ids;
+		std::ranges::transform(values | std::views::take_while(&std::optional<Entity>::has_value), ids.begin(),
+				[this, &insert](auto && entity) {
 					const auto & [entityId, type, value] = *entity;
 					bindMany(insert, 0, entityId, ENTITY_TYPE_VALUES[std::to_underlying(type)], value);
 					if (insert->execute() > 0) {
@@ -293,8 +304,9 @@ namespace WebStat {
 								break;
 						}
 					}
-					existingEntities.emplace(std::get<0>(*entity));
+					return std::get<0>(*entity);
 				});
+		return ids;
 	}
 
 	template<typename... T>
diff --git a/src/ingestor.hpp b/src/ingestor.hpp
index a20071e..8be7360 100644
--- a/src/ingestor.hpp
+++ b/src/ingestor.hpp
@@ -60,13 +60,15 @@ namespace WebStat {
 		size_t linesParsed = 0;
 		size_t linesDiscarded = 0;
 		size_t linesParked = 0;
+		mutable std::flat_set<Crc32Value> existingEntities;
 
 		using JobLastRunTime = std::chrono::system_clock::time_point;
 		JobLastRunTime lastRunIngestParkedLines;
 
 	private:
 		static constexpr size_t MAX_NEW_ENTITIES = 6;
-		void storeEntities(DB::Connection *, std::span<const std::optional<Entity>>) const;
+		using NewEntityIds = std::array<std::optional<Crc32Value>, MAX_NEW_ENTITIES>;
+		NewEntityIds storeEntities(DB::Connection *, std::span<const std::optional<Entity>>) const;
 		using NewEntities = std::array<std::optional<Entity>, MAX_NEW_ENTITIES>;
 		template<typename... T> NewEntities newEntities(const std::tuple<T...> &) const;
 		void handleCurlOperations();
@@ -75,7 +77,6 @@ namespace WebStat {
 		void jobIngestParkedLine(const std::filesystem::path &, uintmax_t size);
 
 		using CurlOperations = std::map<CURL *, std::unique_ptr<CurlOperation>>;
-		mutable std::flat_set<Crc32Value> existingEntities;
 		uint32_t hostnameId;
 		CurlMultiPtr curl;
 		mutable CurlOperations curlOperations;
-- 
cgit v1.2.3