From 5d53704f450788452837b18b26f54d235d883490 Mon Sep 17 00:00:00 2001
From: EuAndreh <eu@euandre.org>
Date: Wed, 30 Oct 2024 20:26:41 -0300
Subject: src/q.go: Fix SQLite ~broken~ transactions

As per expected by SQLite, create new connections so they have
independent transaction, and serialize its use via a channel with a
single consumer.

== Other changes

=== Remove `db` attribute from `queueT` type

Now that `initDB()` can create many database connections (A.K.A. opaque
pointers to the `sqlite3*` object), it makes more sense to hand to it
the responsability to create and destroy these databases.  So now a
`queries.close()` includes what previously was `db.Close()`, and needing
to do it was the only reason that made us keep the `db` attribute.

The arguments to `initDB()` were adjusted to reflect that, as it no
longer is given an initialized database handle, but only the database
path, and does the creation of the handles by itself.

=== Shoehorn multi-statement queries into single-statement ones

In order to avoid creating one `execSerialized()` function per attribute
of the `queryT` queries, we try to leverage SQLite's built-in
transactionality per individual statement. So compound queries that were
previously done in multiple statements wrapped with a `inTx()` that
could be shoe-horned into a single `SELECT` were rewritten to avoid
having more of this application-level serialization.

The topic/consumer owner validation was also changed: now it is a
trigger on the relevant tables that `ABORT` the implicit/invisible
transaction and makes the `INSERT` fail.

In order to make this possible, the `"%s_owners"` table now has an extra
column: `instance_id`.  Despite being extra data, this meta-information
isn't duplicated from anywhere else, and it is an actual useful
information for operators to leverage.

So now the trigger is responsible for stopping the transaction from
going forward without adding an explicit `inTx()` around it, and the
writes can mostly be shrunk to single-statement queries.

Fortunately, all this is enough to fix the `new-instance-takeover`
functional test.
---
 tests/queries.sql | 107 ++++++++++++++++++++++++++++++++++++++++--------------
 1 file changed, 80 insertions(+), 27 deletions(-)

(limited to 'tests/queries.sql')

diff --git a/tests/queries.sql b/tests/queries.sql
index c821e25..e790d41 100644
--- a/tests/queries.sql
+++ b/tests/queries.sql
@@ -27,6 +27,7 @@
 			consumer     TEXT    NOT NULL,
 			message_id   INTEGER NOT NULL
 				REFERENCES "q_messages"(id),
+			instance_id  INTEGER NOT NULL,
 			UNIQUE (consumer, message_id)
 		) STRICT;
 		CREATE INDEX IF NOT EXISTS "q_offsets_consumer"
@@ -38,6 +39,7 @@
 			consumer     TEXT    NOT NULL,
 			message_id   INTEGER NOT NULL
 				REFERENCES "q_messages"(id),
+			instance_id  INTEGER NOT NULL,
 			UNIQUE (consumer, message_id)
 		) STRICT;
 		CREATE INDEX IF NOT EXISTS "q_deadletters_consumer"
@@ -58,6 +60,44 @@
 			owner_id     INTEGER NOT NULL,
 			UNIQUE (topic, consumer)
 		) STRICT;
+
+		CREATE TRIGGER IF NOT EXISTS "q_check_instance_owns_topic"
+			BEFORE INSERT ON "q_offsets"
+			WHEN NEW.instance_id != (
+				SELECT owner_id FROM "q_owners"
+				WHERE topic = (
+					SELECT "q_payloads".topic
+					FROM "q_payloads"
+					JOIN "q_messages" ON "q_payloads".id =
+						"q_messages".payload_id
+					WHERE "q_messages".id = NEW.message_id
+				) AND consumer = NEW.consumer
+			)
+		BEGIN
+			SELECT RAISE(
+				ABORT,
+				'instance does not own topic/consumer combo'
+			);
+		END;
+
+		CREATE TRIGGER IF NOT EXISTS "q_check_can_publish_deadletter"
+			BEFORE INSERT ON "q_deadletters"
+			WHEN NEW.instance_id != (
+				SELECT owner_id FROM "q_owners"
+				WHERE topic = (
+					SELECT "q_payloads".topic
+					FROM "q_payloads"
+					JOIN "q_messages" ON "q_payloads".id =
+						"q_messages".payload_id
+					WHERE "q_messages".id = NEW.message_id
+				) AND consumer = NEW.consumer
+			)
+		BEGIN
+			SELECT RAISE(
+				ABORT,
+				'Instance does not own topic/consumer combo'
+			);
+		END;
 	
 
 --    read:
@@ -81,7 +121,6 @@
 		INSERT INTO "q_payloads" (topic, payload)
 		VALUES (?, ?);
 
-		-- FIXME: must be inside a trnsaction
 		INSERT INTO "q_messages" (uuid, flow_id, payload_id)
 		VALUES (?, ?, last_insert_rowid());
 	
@@ -112,6 +151,38 @@
 		LIMIT 1;
 	
 
+--   owner:
+
+-- next.sql:
+--   write:
+
+--    read:
+		SELECT
+			(
+				SELECT owner_id FROM "q_owners"
+				WHERE
+					topic    = ? AND
+					consumer = ?
+					LIMIT 1
+			) AS owner_id,
+			"q_messages".id,
+			"q_messages".timestamp,
+			"q_messages".uuid,
+			"q_messages".flow_id,
+			"q_payloads".payload
+		FROM "q_messages"
+		JOIN "q_payloads" ON
+			"q_payloads".id = "q_messages".payload_id
+		WHERE
+			"q_payloads".topic = ? AND
+			"q_messages".id NOT IN (
+				SELECT message_id FROM "q_offsets"
+				WHERE consumer = ?
+			)
+		ORDER BY "q_messages".id ASC
+		LIMIT 1;
+	
+
 --   owner:
 
 -- pending.sql:
@@ -146,46 +217,28 @@
 
 -- commit.sql:
 --   write:
-		INSERT INTO "q_offsets" (consumer, message_id)
-		VALUES (?, (SELECT id FROM "q_messages" WHERE uuid = ?));
+		INSERT INTO "q_offsets" (consumer, message_id, instance_id)
+		VALUES (?, (SELECT id FROM "q_messages" WHERE uuid = ?), ?);
 	
 
 --    read:
-		SELECT "q_payloads".topic from "q_payloads"
-		JOIN "q_messages" ON
-			"q_payloads".id = "q_messages".payload_id
-		WHERE "q_messages".uuid = ?;
-	
 
 --   owner:
-		SELECT owner_id FROM "q_owners"
-		WHERE
-			topic    = ? AND
-			consumer = ?;
-	
 
 -- toDead.sql:
 --   write:
-		INSERT INTO "q_offsets"     (      consumer, message_id)
-		VALUES (   ?, (SELECT id FROM "q_messages" WHERE uuid = ?));
+		INSERT INTO "q_offsets"
+			(      consumer, message_id, instance_id)
+		VALUES (   ?, (SELECT id FROM "q_messages" WHERE uuid = ?), ?);
 
-		INSERT INTO "q_deadletters" (uuid, consumer, message_id)
-		VALUES (?, ?, (SELECT id FROM "q_messages" WHERE uuid = ?));
+		INSERT INTO "q_deadletters"
+			(uuid, consumer, message_id, instance_id)
+		VALUES (?, ?, (SELECT id FROM "q_messages" WHERE uuid = ?), ?);
 	
 
 --    read:
-		SELECT "q_payloads".topic FROM "q_payloads"
-		JOIN "q_messages" ON
-			"q_payloads".id = "q_messages".payload_id
-		WHERE "q_messages".uuid = ?;
-	
 
 --   owner:
-		SELECT owner_id FROM "q_owners"
-		WHERE
-			topic    = ? AND
-			consumer = ?;
-	
 
 -- replay.sql:
 --   write:
-- 
cgit v1.2.3

