diff --git a/.e2e-ref b/.e2e-ref index 2110390..90e17f6 100644 --- a/.e2e-ref +++ b/.e2e-ref @@ -1 +1 @@ -be8093f7c76c42cac5b826f40db95db372bdd34b +a183a4f5124d5dc32849852e532987f5be72b88b diff --git a/Resources/Views/self-hosting.leaf b/Resources/Views/self-hosting.leaf index 8603413..645a357 100644 --- a/Resources/Views/self-hosting.leaf +++ b/Resources/Views/self-hosting.leaf @@ -3,7 +3,7 @@ + source-sha256: 9e4558c9375d0b23 -->

Self-hosting

@@ -261,6 +261,32 @@ Take a dump first anyway, docker compose exec db pg_dump -U flamenet flamenet, because the reverse is not true, and an older relay against a newer database is not a supported direction.

+

One step by hand for +older SQLite relays

+

This is for a relay on SQLite (no FN_POSTGRES_URL) that +was already running before 25 September 2026. Postgres relays can skip +it, and so can a SQLite relay first started after that date.

+

When email verification was added, the upgrade was meant to mark +every account that already existed as verified, so that switching +verification on later would lock nobody out. On SQLite that step failed +without an error and marked nobody. Nothing changes until you set +FN_EMAIL_VERIFICATION_REQUIRED. Once you do, everyone who +signed up before that upgrade is turned away and asked for a code from +an email that was never sent.

+

The relay now does this correctly, but each upgrade step runs once, +and a relay that already ran the broken one will not run it again. Stop +the relay, copy relay.sqlite somewhere safe (it is wherever +FN_DB points, if you set it), open it with +sqlite3 relay.sqlite, and run:

+
UPDATE users SET email_verified_at = CURRENT_TIMESTAMP
+WHERE email_verified_at IS NULL
+  AND (created_at IS NULL OR created_at <= (
+    SELECT created_at FROM _fluent_migrations
+    WHERE name = &#("#")39;FlamenetRelay.AddEmailVerification&#("#")39;));
+

It marks only the accounts that existed before verification arrived. +Anyone who signed up since still has to confirm their address. Running +it a second time, or on a relay that never needed it, changes +nothing.

Telling your users about updates

The Android app asks your relay whether a newer build diff --git a/Sources/FlamenetRelay/Retention.swift b/Sources/FlamenetRelay/Retention.swift index 49022ee..361317d 100644 --- a/Sources/FlamenetRelay/Retention.swift +++ b/Sources/FlamenetRelay/Retention.swift @@ -120,7 +120,7 @@ struct Retention: @unchecked Sendable { guard seen < deviceCutoff, seen < newestPerUser[device.userID] ?? .distantPast else { continue } - try await db.transaction { tx in + try await db.retryingTransaction { tx in try await DeviceRetirement.retire( device, because: .expired, audience: config.audience, on: tx) } diff --git a/Sources/FlamenetRelay/Routes.swift b/Sources/FlamenetRelay/Routes.swift index 7a55613..6ba121b 100644 --- a/Sources/FlamenetRelay/Routes.swift +++ b/Sources/FlamenetRelay/Routes.swift @@ -187,7 +187,7 @@ struct Routes: Sendable { throw WireError.tokenScopedElsewhere } - try await request.db.transaction { db in + try await request.db.retryingTransaction { db in let existing = try await Device.find(input.device_id, on: db) // A device id belongs to exactly one account, forever. if let existing, existing.userID != caller.userID { @@ -452,7 +452,7 @@ struct Routes: Sendable { // Its prekey pools and mailbox go with it, or the rows outlive the // identity they were published under. - try await request.db.transaction { db in + try await request.db.retryingTransaction { db in try await DeviceRetirement.retire( device, because: .revoked, audience: self.config.audience, on: db) } @@ -636,7 +636,7 @@ struct Routes: Sendable { request.logger.notice("\(detail)") } - let ids: [Int] = try await request.db.transaction { db in + let ids: [Int] = try await request.db.retryingTransaction { db in var written: [Int] = [] for m in input.messages where owned.contains(m.to_device) { let e = Envelope() @@ -913,7 +913,7 @@ struct Routes: Sendable { throw WireError.rateLimited( "Too many prekey uploads. Try again shortly.", retryAfter: wait) } - try await request.db.transaction { db in + try await request.db.retryingTransaction { db in try await Self.insertPreKeys( input.prekeys, input.pq_prekeys, device: input.device_id, on: db) diff --git a/Sources/FlamenetRelay/RoutesDeliveryTokens.swift b/Sources/FlamenetRelay/RoutesDeliveryTokens.swift index 7f44026..832d7bf 100644 --- a/Sources/FlamenetRelay/RoutesDeliveryTokens.swift +++ b/Sources/FlamenetRelay/RoutesDeliveryTokens.swift @@ -54,7 +54,7 @@ extension Routes { "Too many delivery-token updates. Try again shortly.", retryAfter: wait) } - try await request.db.transaction { db in + try await request.db.retryingTransaction { db in // Revocation first. if !dropping.isEmpty { try await DeliveryToken.query(on: db) diff --git a/Sources/FlamenetRelay/RoutesFederation.swift b/Sources/FlamenetRelay/RoutesFederation.swift index e4879ed..48ea1c4 100644 --- a/Sources/FlamenetRelay/RoutesFederation.swift +++ b/Sources/FlamenetRelay/RoutesFederation.swift @@ -7,12 +7,19 @@ extension Routes { /// Resolve an fnid to a local account, or 404. private func localUser(forFNID fnid: String, on db: any Database) async throws -> User { - guard let user = try await User.query(on: db).filter(\.$fnid == fnid).first() else { + guard let user = try await Self.findLocalUser(forFNID: fnid, on: db) else { throw WireError.notFound("no such identity here") } return user } + /// The local account for an fnid, or nil when there is none. A storage + /// failure throws: "no such account" and "could not look" are different + /// answers, and a peer is told only the first. + static func findLocalUser(forFNID fnid: String, on db: any Database) async throws -> User? { + try await User.query(on: db).filter(\.$fnid == fnid).first() + } + private func verifyPeer(_ request: Request, body: Data, path: String, now: Int) async throws -> Federation.Peer { @@ -58,42 +65,8 @@ extension Routes { throw WireError.rateLimited("peer \(peer.audience) is over its hourly budget here") } - let (accepted, unknown) = try await request.db.transaction { - db -> (Int, [String]) in - var accepted = 0 - var unknown: [String] = [] - for e in input.envelopes { - guard let user = try? await self.localUser(forFNID: e.to_fnid, on: db), - let userID = user.id - else { - // Named, not silently dropped: a peer holding an envelope - // it cannot place should stop retrying it, and it can only - // know that if we say so. - unknown.append(e.to_fnid) - continue - } - // The device must belong to that account. - guard let device = try await Device.find(e.to_device, on: db), - device.userID == userID - else { - unknown.append(e.to_fnid) - continue - } - - let row = Envelope() - row.toUser = userID - row.toDevice = e.to_device - // Sealed by construction across a federation boundary: we were - // not told who sent it and must not invent an attribution. - row.fromUser = 0 - row.fromDevice = "" - row.payload = e.payload - row.envelopeID = e.envelope_id - row.delivered = false - try await row.create(on: db) - accepted += 1 - } - return (accepted, unknown) + let (accepted, unknown) = try await request.db.retryingTransaction { db in + try await Self.placeDeposits(input.envelopes, on: db) } return try .json( @@ -101,6 +74,55 @@ extension Routes { ok: true, accepted: accepted, unknown: unknown)) } + /// Put a peer's envelopes in the local mailboxes they name, and say which + /// recipients this relay does not have. + /// + /// "Unknown" is an answer the peer acts on: it stops retrying that + /// envelope. So it is only ever said about a recipient that is really not + /// here. This used to look the account up with `try?`, which turned a + /// storage failure (a locked SQLite file, a dropped connection) into + /// "unknown", and the peer then gave up on a message this relay would have + /// taken a moment later. A failure throws now, the whole batch rolls back, + /// and the peer gets a 500 and tries again. + static func placeDeposits( + _ envelopes: [FederationDepositRequest.Item], on db: any Database + ) async throws -> (accepted: Int, unknown: [String]) { + var accepted = 0 + var unknown: [String] = [] + for e in envelopes { + guard let user = try await findLocalUser(forFNID: e.to_fnid, on: db), + let userID = user.id + else { + // Named, not silently dropped: a peer holding an envelope + // it cannot place should stop retrying it, and it can only + // know that if we say so. + unknown.append(e.to_fnid) + continue + } + // The device must belong to that account. + guard let device = try await Device.find(e.to_device, on: db), + device.userID == userID + else { + unknown.append(e.to_fnid) + continue + } + + let row = Envelope() + row.toUser = userID + row.toDevice = e.to_device + // Sealed by construction across a federation boundary: we were + // not told who sent it and must not invent an attribution. + row.fromUser = 0 + row.fromDevice = "" + row.payload = e.payload + row.envelopeID = e.envelope_id + row.delivered = false + try await row.create(on: db) + accepted += 1 + } + return (accepted, unknown) + } + /// A batch cap, so one request cannot be an unbounded write. static let maxFederatedBatch = 64 diff --git a/Sources/FlamenetRelay/RoutesGroups.swift b/Sources/FlamenetRelay/RoutesGroups.swift index 92e3c9f..e2544f3 100644 --- a/Sources/FlamenetRelay/RoutesGroups.swift +++ b/Sources/FlamenetRelay/RoutesGroups.swift @@ -24,7 +24,7 @@ extension Routes { try await requireReachable(members, on: request.db) let gid = randomHexToken() - try await request.db.transaction { db in + try await request.db.retryingTransaction { db in let group = Group() group.id = gid group.ownerUserID = caller.userID @@ -91,7 +91,7 @@ extension Routes { throw WireError.badRequest("Groups are capped at \(Group.maxMembers) members.") } - try await request.db.transaction { db in + try await request.db.retryingTransaction { db in for uid in adding.sorted() { let m = GroupMember() m.gid = gid @@ -127,7 +127,7 @@ extension Routes { throw WireError.badRequest("Not a member.") } - try await request.db.transaction { db in + try await request.db.retryingTransaction { db in try await GroupMember.query(on: db) .filter(\.$gid == gid).filter(\.$userID == caller.userID).delete() diff --git a/Sources/FlamenetRelay/RoutesMobile.swift b/Sources/FlamenetRelay/RoutesMobile.swift index d88dd0b..bab5390 100644 --- a/Sources/FlamenetRelay/RoutesMobile.swift +++ b/Sources/FlamenetRelay/RoutesMobile.swift @@ -111,7 +111,7 @@ struct MobileRoutes: Sendable { // Replaced whole, in one transaction, or a failure half way through // leaves somebody's list of who may see them in a state neither they // nor this relay chose. - try await request.db.transaction { db in + try await request.db.retryingTransaction { db in try await RosterTag.query(on: db).filter(\.$userID == me).delete() for digest in digests { let row = RosterTag() diff --git a/Sources/FlamenetRelay/RoutesSealedGroups.swift b/Sources/FlamenetRelay/RoutesSealedGroups.swift index c61f14d..9d5d499 100644 --- a/Sources/FlamenetRelay/RoutesSealedGroups.swift +++ b/Sources/FlamenetRelay/RoutesSealedGroups.swift @@ -63,7 +63,7 @@ extension Routes { gid: gid, version: 1, blob: input.roster_blob) else { throw WireError.forbidden("Roster signature did not verify.") } - try await request.db.transaction { db in + try await request.db.retryingTransaction { db in let group = existing ?? Group() group.id = gid // Zeroed. @@ -158,7 +158,7 @@ extension Routes { } } - try await request.db.transaction { db in + try await request.db.retryingTransaction { db in group.rosterBlob = input.roster_blob group.rosterVersion = input.roster_version if let next = input.new_owner_key { group.ownerKey = next } @@ -185,7 +185,7 @@ extension Routes { let input = try decodeBody(LeaveSealedGroupRequest.self, from: body) guard let tag = validTag(input.tag) else { throw WireError.badRequest("Bad tag.") } - try await request.db.transaction { db in + try await request.db.retryingTransaction { db in try await GroupTag.query(on: db) .filter(\.$gid == gid).filter(\.$tag == tag).delete() // The last one out deletes the group rather than leaving a blob diff --git a/Sources/FlamenetRelay/SQLiteBusyRetry.swift b/Sources/FlamenetRelay/SQLiteBusyRetry.swift new file mode 100644 index 0000000..1286112 --- /dev/null +++ b/Sources/FlamenetRelay/SQLiteBusyRetry.swift @@ -0,0 +1,69 @@ +import Fluent +import Foundation +import SQLiteNIO + +/// A transaction that SQLite refused with "database is locked" runs again. +/// +/// SQLiteNIO installs a busy handler that waits for a lock however long it +/// takes, so it looks as if a relay on SQLite can never see SQLITE_BUSY. It +/// can. Fluent opens every transaction with a plain `BEGIN`, which takes no +/// lock until the first statement, and a transaction whose first statement is +/// a read holds a shared lock when it reaches its first write. If another +/// connection is writing at that moment, each is waiting on the other, and +/// SQLite answers the reader with SQLITE_BUSY at once rather than calling the +/// busy handler, because waiting would never end. +/// +/// That is what the conformance job kept losing to (relay!165, 2026-09-25): +/// the retention sweep starts at boot and deletes, the first test registers a +/// device, and registration opens with `Device.find` before it writes. The +/// phone on the other end sees a 500 "Storage failure." for a request that +/// would have succeeded a few milliseconds later, and a self-hosted relay on +/// SQLite hits the same thing whenever two writes land together. +/// +/// Rolling back releases the reader's lock, which is what lets the writer +/// finish, so the retry succeeds once that commit is through. It is only safe +/// because every closure passed here touches nothing but the database: a +/// rolled-back attempt leaves no trace, and running it again is the same as +/// running it once. A closure that sends, wakes or deletes a file does not +/// belong in here. +/// +/// Postgres never throws `SQLiteError`, so on Postgres this is `transaction`. +extension Database { + func retryingTransaction( + attempts: Int = SQLiteBusy.attempts, + _ closure: @escaping @Sendable (any Database) async throws -> T + ) async throws -> T { + var attempt = 1 + while true { + do { + return try await transaction(closure) + } catch let error as SQLiteError where SQLiteBusy.isBusy(error) && attempt < attempts { + logger.notice("sqlite busy on attempt \(attempt), retrying the transaction") + try await Task.sleep(nanoseconds: SQLiteBusy.backoff(attempt: attempt)) + attempt += 1 + } + } + } +} + +enum SQLiteBusy { + /// Five tries is about a second of waiting in total. A deadlock clears as + /// soon as the other side commits, so more than that means something is + /// holding the database for real, and the caller should hear about it. + static let attempts = 5 + + static func isBusy(_ error: SQLiteError) -> Bool { + switch error.reason { + // busyInSnapshot is the same deadlock on a relay whose operator has + // switched the file to WAL. + case .busy, .busyInSnapshot: return true + default: return false + } + } + + /// 25 ms, 50, 75, 100, each plus up to 25 ms of jitter so two requests + /// that collided once do not collide again on the same schedule. + static func backoff(attempt: Int) -> UInt64 { + UInt64(attempt) * 25_000_000 + UInt64.random(in: 0..<25_000_000) + } +} diff --git a/Tests/FlamenetRelayTests/FederatedUnknownMeansAbsentTests.swift b/Tests/FlamenetRelayTests/FederatedUnknownMeansAbsentTests.swift new file mode 100644 index 0000000..158c341 --- /dev/null +++ b/Tests/FlamenetRelayTests/FederatedUnknownMeansAbsentTests.swift @@ -0,0 +1,78 @@ +import Fluent +import Foundation +import SQLKit +import Testing +import Vapor + +@testable import FlamenetRelay + +/// A federated deposit reports a recipient "unknown" only when this relay +/// really does not have them. +/// +/// The peer acts on that word: it stops retrying the envelope. The lookup used +/// to run under `try?`, so a storage failure read as "unknown" and the peer +/// gave up on a message that would have been accepted a moment later. Found +/// while fixing the SQLite lock race (SQLiteBusyRetry.swift), which is exactly +/// the kind of passing failure this turned into a lost message. +@Suite("a federated recipient is unknown only when absent") +struct FederatedUnknownMeansAbsentTests { + + private func item(_ fnid: String, device: String = "bbbb-0002") -> FederationDepositRequest.Item { + FederationDepositRequest.Item( + to_fnid: fnid, to_device: device, payload: "from-afar", envelope_id: nil) + } + + private func addBob(_ h: RelayHarness) async throws { + try await h.register(h.device(user: 2, id: "bbbb-0002"), on: h.db) + let user = User() + user.id = 2 + user.username = "bob" + user.email = "bob@example.com" + user.passwordHash = "x" + user.displayName = "Bob" + user.chosenStatus = ChosenStatus.online.rawValue + user.fnid = "fnid:bob" + try await user.create(on: h.db) + } + + @Test("a recipient who is here is accepted") + func presentIsAccepted() async throws { + try await withRelay { h in + try await addBob(h) + let placed = try await Routes.placeDeposits([item("fnid:bob")], on: h.db) + #expect(placed.accepted == 1) + #expect(placed.unknown.isEmpty) + } + } + + /// The control for the test below: absent really is "unknown", so that + /// test cannot pass by never saying the word. + @Test("a recipient who is not here is unknown") + func absentIsUnknown() async throws { + try await withRelay { h in + try await addBob(h) + let placed = try await Routes.placeDeposits([item("fnid:nobody")], on: h.db) + #expect(placed.accepted == 0) + #expect(placed.unknown == ["fnid:nobody"]) + } + } + + @Test("a lookup that fails throws, and is not reported as unknown") + func aFailedLookupIsNotUnknown() async throws { + try await withRelay { h in + try await addBob(h) + // The account lookup now fails for storage reasons, not absence. + let sql = try #require(h.db as? any SQLDatabase) + try await sql.raw("DROP TABLE users").run() + + do { + let placed = try await Routes.placeDeposits([item("fnid:bob")], on: h.db) + Issue.record( + "a storage failure came back as an answer: unknown \(placed.unknown), " + + "so the peer would stop retrying a message this relay could take") + } catch { + // Thrown, so the route answers 500 and the peer retries. + } + } + } +} diff --git a/Tests/FlamenetRelayTests/GrandfatheringWorksOnBothDialectsTests.swift b/Tests/FlamenetRelayTests/GrandfatheringWorksOnBothDialectsTests.swift index 57cfefb..5087264 100644 --- a/Tests/FlamenetRelayTests/GrandfatheringWorksOnBothDialectsTests.swift +++ b/Tests/FlamenetRelayTests/GrandfatheringWorksOnBothDialectsTests.swift @@ -44,7 +44,9 @@ struct GrandfatheringWorksOnBothDialectsTests { /// `username_skeleton`. That is the same trap `AcceptExistingContacts` /// fell into from the other side, and the reason this reads the live /// schema instead of hardcoding a column list that would rot. - private func insertUser(_ app: Application, id: Int, name: String) async throws { + private func insertUser( + _ app: Application, id: Int, name: String, createdAt: Date? = nil + ) async throws { guard let sql = app.db as? any SQLDatabase else { Issue.record("not a SQL database") return @@ -62,7 +64,7 @@ struct GrandfatheringWorksOnBothDialectsTests { ("password_hash", "'x'"), ("display_name", "'\(name.capitalized)'"), ("chosen_status", "'\(ChosenStatus.online.rawValue)'"), - ] + ] + (createdAt.map { [("created_at", "\($0.timeIntervalSince1970)")] } ?? []) let use = candidates.filter { present.contains($0.0) } let columns = use.map(\.0).joined(separator: ", ") let values = use.map(\.1).joined(separator: ", ") @@ -165,4 +167,73 @@ struct GrandfatheringWorksOnBothDialectsTests { + "this test is measuring the wrong thing" #expect(refused, "\(wrongThing)") } + + /// A relay that already ran the broken migration will not run the fixed + /// one, so /self-hosting prints an UPDATE for the operator to run by hand. + /// This runs the SQL exactly as the page prints it, read out of the + /// rendered page, against a database left in the state the broken + /// migration left: an account from before it unverified, and one from + /// after it that must stay unverified. + @Test("the UPDATE on /self-hosting repairs a relay the broken migration left behind") + func theHandRepairOnTheSelfHostingPageWorks() async throws { + let app = try await Application.make(.testing) + app.logger.logLevel = .critical + app.databases.use(.sqlite(.memory), as: .sqlite) + defer { blockingShutdown(app) } + + let page = URL(fileURLWithPath: #filePath) + .deletingLastPathComponent().deletingLastPathComponent().deletingLastPathComponent() + .appendingPathComponent("Resources/Views/self-hosting.leaf") + let repair = try #require( + Self.sqlBlock(in: try String(contentsOf: page, encoding: .utf8), mentioning: "email_verified_at"), + "no SQL block naming email_verified_at on /self-hosting") + + let all = RelayMigrations.all() + let cut = try #require(all.firstIndex(where: { $0 is AddEmailVerification })) + guard try await migrateUpToEmailVerification(app) else { return } + + try await insertUser(app, id: 1, name: "ada", createdAt: Date().addingTimeInterval(-86_400)) + app.migrations.add(all[cut]) + try await app.autoMigrate() + + let sql = try #require(app.db as? any SQLDatabase) + // What NOW() left behind on SQLite: nobody marked. + try await sql.raw("UPDATE users SET email_verified_at = NULL").run() + try await insertUser(app, id: 2, name: "bob", createdAt: Date().addingTimeInterval(3_600)) + + // The rest of the migrations, so the UPDATE meets today's schema. + for m in all[(cut + 1)...] { app.migrations.add(m) } + try await app.autoMigrate() + + try await sql.raw(SQLQueryString(repair)).run() + // Twice, because the page says a second run changes nothing. + try await sql.raw(SQLQueryString(repair)).run() + + // Through the model, which is how the relay reads it: a value the + // model could not decode would lock the account out a different way. + let ada = try #require(try await User.find(1, on: app.db)) + let bob = try #require(try await User.find(2, on: app.db)) + #expect(ada.emailVerifiedAt != nil, "the account from before the migration is still locked out") + #expect(bob.emailVerifiedAt == nil, "the repair verified an account that signed up afterwards") + } + + /// The text of the first `

` block containing `needle`,
+    /// with pandoc's HTML escaping and Leaf's `#` escaping undone.
+    static func sqlBlock(in page: String, mentioning needle: String) -> String? {
+        var rest = page[...]
+        while let open = rest.range(of: "
") {
+            guard let close = rest.range(of: "
", range: open.upperBound..")
+                .replacingOccurrences(of: """, with: "\"")
+                .replacingOccurrences(of: "'", with: "'")
+                .replacingOccurrences(of: "&", with: "&")
+            if body.contains(needle) { return body }
+            rest = rest[close.upperBound...]
+        }
+        return nil
+    }
 }
diff --git a/Tests/FlamenetRelayTests/PoolerCompatibilityTests.swift b/Tests/FlamenetRelayTests/PoolerCompatibilityTests.swift
index 50a9dab..54c6763 100644
--- a/Tests/FlamenetRelayTests/PoolerCompatibilityTests.swift
+++ b/Tests/FlamenetRelayTests/PoolerCompatibilityTests.swift
@@ -105,7 +105,7 @@ final class PoolerCompatibilityTests: XCTestCase {
         // And it can find something it is not looking for, so the matching
         // itself works.
         XCTAssertTrue(
-            files.contains { $0.body.contains("db.transaction") },
+            files.contains { Self.opensATransaction($0.body) },
             "the scan cannot see into the files it is reading")
     }
 
@@ -170,11 +170,19 @@ final class PoolerCompatibilityTests: XCTestCase {
                 + "than be tuned")
     }
 
+    /// Either spelling. Routes open them through `retryingTransaction` since
+    /// 2026-09-26 (see SQLiteBusyRetry.swift), which is `transaction` with a
+    /// retry on SQLite and exactly `transaction` on Postgres, so nothing about
+    /// pooling changes.
+    private static func opensATransaction(_ body: String) -> Bool {
+        body.contains("db.transaction") || body.contains("db.retryingTransaction")
+    }
+
     /// Transactions are FINE, and this records that rather than leaving it to
     /// be re-derived.
     func testTransactionsAreExpectedAndAllowed() throws {
         let files = try everySource()
-        let users = files.filter { $0.body.contains("db.transaction") }
+        let users = files.filter { Self.opensATransaction($0.body) }
         XCTAssertGreaterThan(
             users.count, 3,
             "the relay stopped using transactions, which is a "
diff --git a/Tests/FlamenetRelayTests/SQLiteBusyRetryTests.swift b/Tests/FlamenetRelayTests/SQLiteBusyRetryTests.swift
new file mode 100644
index 0000000..54b0893
--- /dev/null
+++ b/Tests/FlamenetRelayTests/SQLiteBusyRetryTests.swift
@@ -0,0 +1,190 @@
+import Fluent
+import Foundation
+import NIOPosix
+import SQLKit
+import SQLiteNIO
+import Testing
+import Vapor
+
+@testable import FlamenetRelay
+
+/// A transaction that reads before it writes can lose to another writer on
+/// SQLite, and when it does it is run again rather than answered with a 500.
+///
+/// The conformance job failed on this (relay!165, 2026-09-25): the first
+/// device registration of the run came back "Storage failure." because the
+/// retention sweep was deleting at the same moment. The busy handler SQLiteNIO
+/// installs does not help, since SQLite skips it for exactly this shape.
+///
+/// These tests build the collision on purpose with two real connections to one
+/// file, rather than hoping two requests happen to overlap. The first test
+/// shows the collision is real with a plain `transaction`, so that the second
+/// passing means the retry got through it and not that nothing collided.
+@Suite("a transaction that loses a lock race on SQLite is run again")
+struct SQLiteBusyRetryTests {
+
+    /// One file, two connections. Fluent keeps one SQLite connection per event
+    /// loop, so two loops is two connections. Four threads in the pool
+    /// because the losing writer's COMMIT spins on a pool thread until the
+    /// reader lets go, and the reader needs a thread of its own to do that.
+    private func withTwoConnections(
+        _ body: (any Database, any Database) async throws -> Void
+    ) async throws {
+        let loops = MultiThreadedEventLoopGroup(numberOfThreads: 2)
+        let pool = NIOThreadPool(numberOfThreads: 4)
+        pool.start()
+        let app = try await Application.make(.testing, .shared(loops))
+        app.logger.logLevel = .critical
+        app.threadPool = pool
+        let path = FileManager.default.temporaryDirectory
+            .appendingPathComponent("busy-\(UUID().uuidString).sqlite").path
+        app.databases.use(.sqlite(.file(path)), as: .sqlite)
+
+        var failure: (any Error)?
+        do {
+            let el = Array(loops.makeIterator())
+            let a = try #require(app.databases.database(.sqlite, logger: app.logger, on: el[0]))
+            let b = try #require(app.databases.database(.sqlite, logger: app.logger, on: el[1]))
+            try await sql(a).raw("CREATE TABLE t (x INTEGER)").run()
+            try await body(a, b)
+        } catch {
+            failure = error
+        }
+        try? await app.asyncShutdown()
+        try? await pool.shutdownGracefully()
+        try? await loops.shutdownGracefully()
+        try? FileManager.default.removeItem(atPath: path)
+        if let failure { throw failure }
+    }
+
+    private func sql(_ db: any Database) -> any SQLDatabase {
+        db as! any SQLDatabase
+    }
+
+    /// A reads, B writes and so holds the write lock, then A writes. A is
+    /// holding a read lock B needs released before it can commit, and B is
+    /// holding the write lock A needs, so SQLite refuses A at once.
+    private func collide(
+        a: any Database, b: any Database, retrying: Bool, attempts: Counter
+    ) async throws {
+        let baton = Baton()
+        let writer = Task {
+            try await b.transaction { tx in
+                try await baton.wait("a read")
+                try await self.sql(tx).raw("INSERT INTO t (x) VALUES (2)").run()
+                await baton.raise("b wrote")
+            }
+        }
+        let reader: @Sendable (any Database) async throws -> Void = { tx in
+            await attempts.bump()
+            _ = try await self.sql(tx).raw("SELECT count(*) AS n FROM t").all()
+            await baton.raise("a read")
+            try await baton.wait("b wrote")
+            try await self.sql(tx).raw("INSERT INTO t (x) VALUES (1)").run()
+        }
+        do {
+            if retrying {
+                try await a.retryingTransaction(reader)
+            } else {
+                try await a.transaction(reader)
+            }
+        } catch {
+            // The writer commits once the reader has rolled back; wait for it
+            // either way so the next test does not inherit a held lock.
+            try? await writer.value
+            throw error
+        }
+        try await writer.value
+    }
+
+    @Test("a plain transaction loses: SQLite answers busy, not wait")
+    func theCollisionIsReal() async throws {
+        try await withTwoConnections { a, b in
+            let attempts = Counter()
+            do {
+                try await collide(a: a, b: b, retrying: false, attempts: attempts)
+                Issue.record("no collision, so the retry test below proves nothing")
+            } catch let e as SQLiteError {
+                #expect(SQLiteBusy.isBusy(e), "failed, but not with busy: \(e)")
+            }
+            #expect(await attempts.value == 1)
+        }
+    }
+
+    @Test("the retrying transaction gets through and both rows land")
+    func theRetryGetsThrough() async throws {
+        try await withTwoConnections { a, b in
+            let attempts = Counter()
+            try await collide(a: a, b: b, retrying: true, attempts: attempts)
+
+            // More than one attempt, or the collision did not happen and this
+            // passed for free.
+            #expect(await attempts.value > 1)
+            let rows = try await sql(a).raw("SELECT x FROM t ORDER BY x").all()
+                .map { try $0.decode(column: "x", as: Int.self) }
+            // Exactly once each: the rolled-back attempt left nothing behind.
+            #expect(rows == [1, 2])
+        }
+    }
+
+    @Test("any other failure is thrown on the first attempt, not retried")
+    func otherErrorsAreNotRetried() async throws {
+        try await withTwoConnections { a, _ in
+            let attempts = Counter()
+            do {
+                try await a.retryingTransaction { tx in
+                    await attempts.bump()
+                    try await self.sql(tx).raw("INSERT INTO no_such_table VALUES (1)").run()
+                }
+                Issue.record("an insert into a missing table succeeded")
+            } catch let e as SQLiteError {
+                #expect(!SQLiteBusy.isBusy(e))
+            }
+            #expect(await attempts.value == 1)
+        }
+    }
+
+    /// Every transaction the relay opens goes through the retry. A new
+    /// `.transaction {` in a route would bring the 500 back for that route
+    /// and nothing else would say so.
+    @Test("no route opens a transaction that is not retried")
+    func everyTransactionIsRetried() throws {
+        let dir = URL(fileURLWithPath: #filePath)
+            .deletingLastPathComponent().deletingLastPathComponent().deletingLastPathComponent()
+            .appendingPathComponent("Sources/FlamenetRelay")
+        let files = try FileManager.default.contentsOfDirectory(atPath: dir.path)
+            .filter { $0.hasSuffix(".swift") }
+        #expect(files.count > 20, "read \(files.count) source files, so the scan below means little")
+        var offenders: [String] = []
+        for name in files {
+            let text = try String(contentsOf: dir.appendingPathComponent(name), encoding: .utf8)
+            for (i, line) in text.split(separator: "\n", omittingEmptySubsequences: false).enumerated()
+            where line.contains(".transaction {") || line.contains(".transaction(") {
+                if line.trimmingCharacters(in: .whitespaces).hasPrefix("//") { continue }
+                offenders.append("\(name):\(i + 1)")
+            }
+        }
+        #expect(offenders.isEmpty, "use retryingTransaction: \(offenders)")
+    }
+}
+
+private actor Baton {
+    private var raised: Set = []
+
+    func raise(_ name: String) { raised.insert(name) }
+
+    func wait(_ name: String) async throws {
+        let deadline = Date().addingTimeInterval(10)
+        while !raised.contains(name) {
+            guard Date() < deadline else {
+                throw Abort(.internalServerError, reason: "timed out waiting for \(name)")
+            }
+            try await Task.sleep(nanoseconds: 5_000_000)
+        }
+    }
+}
+
+private actor Counter {
+    private(set) var value = 0
+    func bump() { value += 1 }
+}