.e2e-ref

@@ -1 +1 @@
-be8093f7c76c42cac5b826f40db95db372bdd34b
+a183a4f5124d5dc32849852e532987f5be72b88b

Resources/Views/self-hosting.leaf

@@ -3,7 +3,7 @@
 <!-- GENERATED by Scripts/build-doc-pages.sh from flamenet-e2e/SELF-HOSTING.md.
      Edits made HERE are deleted the next time that script runs.
      Change the markdown source and re-run it.
-     source-sha256: 3007e08fbe70bd93 -->
+     source-sha256: 9e4558c9375d0b23 -->
 <article class="band doc doc-generated">
   <div class="wrap wrap-narrow">
 <h1 id="self-hosting">Self-hosting</h1>
@@ -261,6 +261,32 @@ Take a dump first anyway,
 <code>docker compose exec db pg_dump -U flamenet flamenet</code>,
 because the reverse is not true, and an older relay against a newer
 database is not a supported direction.</p>
+<h3 id="one-step-by-hand-for-older-sqlite-relays">One step by hand for
+older SQLite relays</h3>
+<p>This is for a relay on SQLite (no <code>FN_POSTGRES_URL</code>) that
+was already running before 25 September 2026. Postgres relays can skip
+it, and so can a SQLite relay first started after that date.</p>
+<p>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
+<code>FN_EMAIL_VERIFICATION_REQUIRED</code>. 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.</p>
+<p>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 <code>relay.sqlite</code> somewhere safe (it is wherever
+<code>FN_DB</code> points, if you set it), open it with
+<code>sqlite3 relay.sqlite</code>, and run:</p>
+<pre class="sql"><code>UPDATE users SET email_verified_at = CURRENT_TIMESTAMP
+WHERE email_verified_at IS NULL
+  AND (created_at IS NULL OR created_at &lt;= (
+    SELECT created_at FROM _fluent_migrations
+    WHERE name = &#("#")39;FlamenetRelay.AddEmailVerification&#("#")39;));</code></pre>
+<p>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.</p>
 <h2 id="telling-your-users-about-updates">Telling your users about
 updates</h2>
 <p>The Android app asks <em>your</em> relay whether a newer build

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)
                 }

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)

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)

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
 

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()
 

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()

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

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<T: Sendable>(
+        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)
+    }
+}

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.
+            }
+        }
+    }
+}

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 `<pre class="sql">` 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: "<pre class=\"sql\"><code>") {
+            guard let close = rest.range(of: "</code></pre>", range: open.upperBound..<rest.endIndex)
+            else { return nil }
+            let body = String(rest[open.upperBound..<close.lowerBound])
+                .replacingOccurrences(of: "#(\"#\")", with: "#")
+                .replacingOccurrences(of: "&lt;", with: "<")
+                .replacingOccurrences(of: "&gt;", with: ">")
+                .replacingOccurrences(of: "&quot;", with: "\"")
+                .replacingOccurrences(of: "&#39;", with: "'")
+                .replacingOccurrences(of: "&amp;", with: "&")
+            if body.contains(needle) { return body }
+            rest = rest[close.upperBound...]
+        }
+        return nil
+    }
 }

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 "

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<String> = []
+
+    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 }
+}