extant2000/flamenet-relay
1
0
Fork 0
Code Issues 14 Pull Requests Actions Packages Projects Releases Wiki Activity

Prove delete-on-ack before trusting it, and find it unsafe on SQLite

Browse Source 
The flag deletes ciphertext irrecoverably, so it needed a test that says
WHICH rows go — not that something got deleted. The first run found a real
one.

The cursor is a row id, so "delete everything at or below it" is sound
only where ids are never reused. SQLite hands a deleted rowid straight
back to the next insert: measured, first=1 then late=1, so a message
arriving after the cursor is deleted UNREAD with nothing left to show it
existed. Postgres sequences never reuse. FN_DELETE_ON_ACK is now REFUSED
with a critical log on anything that is not Postgres, rather than
discouraged — a self-hoster on SQLite must not lose mail to a flag.

Five tests, each asserting the survivors and not just the casualties: a
cursor deletes only at or below itself, a later arrival survives when ids
are monotonic, one device's cursor never touches another's inbox, a zero
cursor is not a bound (a reset client would otherwise wipe its own inbox),
and the SQLite reuse itself — so if that ever changes the guard can be
revisited rather than guessed at.

The predicate is extracted as Routes.deleteAcknowledged so the test
exercises the shipped code. A test that restates the query proves only
that it agrees with itself.

Also adds the Redis wake-up layer for held polls: LPUSH a token per
delivered message, BRPOP in the poll. Chosen over Postgres LISTEN because
LISTEN pins a backend per waiter and max_connections here is 100 — a
ceiling of about 86 people with the app open. It is allowed to lose data:
a token only says "look now", the cursor query remains the truth, so
there is no persistence to configure. FN_REDIS_URL unset falls back to the
interval, because a self-hoster should not need two services to run a
messenger.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Claude
2026-08-26 19:05:14 -04:00
parent 8d74042778
commit 9793f93e91
6 changed files with 333 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Package.resolved
+19 -1
View File
@@ -1,5 +1,5 @@
{
"originHash" : "c55e8d1dbb001459cc73385e2f8dd147e4a9a58a932b10d8d3a667f5d91360c0",
"originHash" : "0e96a1a91f14c93c6d07d463090611d9595eff4398110d7d7961469ec42fc59c",
"pins" : [
{
"identity" : "async-http-client",
@@ -109,6 +109,24 @@
"version" : "1.33.1"
}
},
{
"identity" : "redis",
"kind" : "remoteSourceControl",
"location" : "https://github.com/vapor/redis.git",
"state" : {
"revision" : "bc960cc856ef9fd5da760d6a3afedb93fc92aa00",
"version" : "4.14.0"
}
},
{
"identity" : "redistack",
"kind" : "remoteSourceControl",
"location" : "https://github.com/swift-server/RediStack.git",
"state" : {
"revision" : "8c4ce10883e405cdbde52855f97f1ec9468c889b",
"version" : "1.6.3"
}
},
{
"identity" : "routing-kit",
"kind" : "remoteSourceControl",
Package.swift
+4
View File
@@ -17,6 +17,9 @@ let package = Package(
.package(url: "https://github.com/vapor/fluent-postgres-driver.git", from: "2.8.0"),
.package(url: "https://github.com/vapor/leaf.git", from: "4.3.0"),
.package(url: "https://github.com/apple/swift-crypto.git", from: "3.0.0"),
// Wake-up fan-out for long-polling. See RoutesWake.swift — the relay
// works without it, so this is a dependency for latency, not function.
.package(url: "https://github.com/vapor/redis.git", from: "4.11.0"),
],
targets: [
.executableTarget(
@@ -26,6 +29,7 @@ let package = Package(
.product(name: "Fluent", package: "fluent"),
.product(name: "FluentSQLiteDriver", package: "fluent-sqlite-driver"),
.product(name: "FluentPostgresDriver", package: "fluent-postgres-driver"),
.product(name: "Redis", package: "redis"),
.product(name: "Leaf", package: "leaf"),
.product(name: "Crypto", package: "swift-crypto"),
]
Sources/FlamenetRelay/Entrypoint.swift
+40 -2
View File
@@ -4,6 +4,7 @@ import FluentSQLiteDriver
import Foundation
import FluentPostgresDriver
import Leaf
import Redis
import Vapor
/// Flamenet relay — the `/e2e/*` half of the messenger, as a standalone
@@ -597,9 +598,46 @@ func register(routes r: Routes, on app: Application) throws {
}
// Opt-in, because it deletes ciphertext irrecoverably. See
// Routes.DeleteOnAckKey.
// Wake-ups for long-polling. Optional by design: with no Redis the
// held poll falls back to an interval, and the relay is one service
// again. See RoutesWake.swift.
let redisURL = ProcessInfo.processInfo.environment["FN_REDIS_URL"] ?? ""
if !redisURL.isEmpty, let parsed = URL(string: redisURL), let host = parsed.host {
app.redis.configuration = try? RedisConfiguration(
hostname: host,
port: parsed.port ?? 6379,
password: parsed.password,
pool: .init(maximumConnectionCount: .maximumActiveConnections(32)))
app.storage[Wake.EnabledKey.self] = (app.redis.configuration != nil)
if app.storage[Wake.EnabledKey.self] == true {
app.logger.notice("wake-ups via Redis at \(host) — held polls block instead of re-querying")
} else {
app.logger.critical("FN_REDIS_URL is set but unusable; held polls will fall back to polling.")
}
}
let ackFlag = ProcessInfo.processInfo.environment["FN_DELETE_ON_ACK"] ?? ""
app.storage[Routes.DeleteOnAckKey.self] = (ackFlag == "1" || ackFlag.lowercased() == "true")
if app.storage[Routes.DeleteOnAckKey.self] == true {
let ackAsked = (ackFlag == "1" || ackFlag.lowercased() == "true")
// 🔴 **Postgres only, and refused rather than discouraged elsewhere.**
// The cursor is a row id, so "delete everything at or below it" is
// sound only where ids are never reused. SQLite hands a deleted rowid
// straight back to the next insert — measured in DeleteOnAckTests,
// first=1 then late=1 — so the next message arrives already below the
// cursor and is deleted UNREAD, with nothing left to show it existed.
// Postgres sequences never reuse. A self-hoster on SQLite who sets this
// must get a refusal, not silent data loss.
let pgURL = ProcessInfo.processInfo.environment["FN_POSTGRES_URL"] ?? ""
let onPostgres = !pgURL.isEmpty
app.storage[Routes.DeleteOnAckKey.self] = ackAsked && onPostgres
if ackAsked && !onPostgres {
app.logger.critical("""
FN_DELETE_ON_ACK was requested and is REFUSED: this relay is not on Postgres, and a \
database that reuses row ids would delete the next message unread. Polls will mark \
envelopes delivered and let retention sweep them, as before.
""")
} else if ackAsked {
app.logger.notice("FN_DELETE_ON_ACK is on — an advancing poll cursor DELETES the envelopes below it.")
}
e2e.get("messages") { req in
Sources/FlamenetRelay/Routes.swift
+45 -6
View File
@@ -433,11 +433,43 @@ struct Routes: Sendable {
}
return written
}
// ⚠️ **After the transaction, never inside it.** A wake-up sent before
// the commit races the reader: a held poll can be woken, run its cursor
// query against the pre-commit state, find nothing and go back to
// sleep — and the token is already spent, so the message waits for the
// next timeout. Signal only what is durably written.
if !ids.isEmpty {
let woken = Set(input.messages.map(\.to_device)).intersection(owned)
for device in woken {
await Wake.signal(device: device, on: request.application)
}
}
return try .json(SendResponse(ok: true, ids: ids))
}
// MARK: - §8.5 GET /e2e/messages
/// Delete the envelopes a poll cursor has acknowledged.
///
/// Extracted so the test exercises the shipped predicate rather than a
/// copy of it — the whole risk here is *which rows go*, and a test that
/// restates the query proves only that the test agrees with itself.
///
/// - `after == 0` deletes nothing: a client that has ingested nothing has
/// acknowledged nothing, and a cursor reset to zero must not be read as
/// "everything below zero", which is every row for that device.
/// - Scoped to one device. Two devices on one account have independent
/// cursors, and the slower one's mail is not the faster one's to drop.
static func deleteAcknowledged(toDevice deviceID: String, after: Int,
on db: any Database) async throws {
guard after > 0 else { return }
try await Envelope.query(on: db)
.filter(\.$toDevice == deviceID)
.filter(\.$id <= after)
.delete()
}
/// Whether an advancing cursor deletes what it acknowledged. Off unless
/// `FN_DELETE_ON_ACK=1`, because the failure mode is silent data loss.
struct DeleteOnAckKey: StorageKey { typealias Value = Bool }
@@ -477,11 +509,8 @@ struct Routes: Sendable {
// delete: the test device had been reset several times, so an empty
// thread proved nothing either way. Turn it on with a test that sends,
// polls twice, and asserts the row is gone AND the message was read.
if request.application.storage[DeleteOnAckKey.self] == true, after > 0 {
try await Envelope.query(on: request.db)
.filter(\.$toDevice == deviceID)
.filter(\.$id <= after)
.delete()
if request.application.storage[DeleteOnAckKey.self] == true {
try await Self.deleteAcknowledged(toDevice: deviceID, after: after, on: request.db)
}
// How long to hold this request open when there is nothing yet. Zero
@@ -518,7 +547,17 @@ struct Routes: Sendable {
if rows.isEmpty && wait > 0 {
let deadline = Date().addingTimeInterval(Double(wait))
while rows.isEmpty && Date() < deadline {
try? await Task.sleep(nanoseconds: 1_000_000_000)
// Block on a wake-up when there is a Redis to block on, and
// fall back to the interval when there is not. `false` means
// "nothing arrived", not "nothing is there" — a token can be
// missed if Redis restarted or the cap trimmed it — so the
// cursor query below runs either way.
let remaining = max(1, Int(deadline.timeIntervalSinceNow.rounded(.down)))
if await Wake.wait(device: deviceID, timeout: remaining,
on: request.application) == false
&& request.application.storage[Wake.EnabledKey.self] != true {
try? await Task.sleep(nanoseconds: 1_000_000_000)
}
if Task.isCancelled { break }
rows = try await Envelope.query(on: request.db)
.filter(\.$toDevice == deviceID)
Sources/FlamenetRelay/RoutesWake.swift
+83
View File
@@ -0,0 +1,83 @@
import Foundation
import Redis
import Vapor
/// Wake-ups for long-polling.
///
/// A held `GET /e2e/messages` has to learn that something arrived. Three ways
/// to do that, and the reason this one:
///
/// - **Poll the database on an interval.** What the relay does without Redis.
/// Correct, and costs a query per waiter per second.
/// - **Postgres `LISTEN`.** Wakes instantly and **pins a backend connection per
/// waiter**. `max_connections` on this database is 100, so it caps
/// simultaneous waiters at about 86 — not a distant ceiling, roughly 86
/// people with the app open.
/// - **Redis `BRPOP`.** Thousands of blocked clients on one process, and the
/// list holds the token until somebody takes it, so a wake-up that lands
/// between two polls is not lost the way a `NOTIFY` to nobody is.
///
/// 🔴 **This is allowed to lose data, and that is the design.** If Redis
/// restarts, every queued token vanishes — and nothing breaks, because a token
/// only says "look now". The cursor query is what actually returns messages and
/// remains the sole source of truth. So: no persistence to configure, no AOF
/// tuning, no backup story. ⚠️ Do not "fix" this later by turning on durability
/// it does not need.
///
/// ⚠️ The relay runs perfectly well with no Redis at all — `FN_REDIS_URL`
/// unset falls back to the interval. It is a dependency for latency, not for
/// function, and it must stay that way: a self-hoster should not need two
/// services to run a messenger.
enum Wake {
/// Per device, because messages are sealed per device.
static func key(device: String) -> RedisKey { RedisKey("wake:\(device)") }
/// Longest a token is worth keeping. A device that has not polled in a day
/// will do a cursor query when it returns, so the tokens are dead weight.
static let ttlSeconds = 86_400
/// ⚠️ Capped. A device that never comes back would otherwise accumulate one
/// token per message it was sent, forever. Sixty-four is far more than any
/// waiter needs — the first pop sends it to the cursor query, which returns
/// everything regardless of how many tokens are left behind.
static let maxDepth = 63
/// Tell a device something arrived. Best effort by construction.
///
/// ⚠️ **Never throws into the send path.** A message is delivered when the
/// envelope is committed; the wake-up is an optimisation on top. A Redis
/// outage that failed sends would turn a latency dependency into a
/// functional one, which is exactly what the design is avoiding.
static func signal(device: String, on app: Application) async {
guard app.storage[EnabledKey.self] == true else { return }
do {
let k = key(device: device)
_ = try await app.redis.lpush("1", into: k).get()
_ = try await app.redis.ltrim(k, before: 0, after: maxDepth).get()
_ = try await app.redis.expire(k, after: .seconds(Int64(ttlSeconds))).get()
} catch {
app.logger.debug("wake-up for \(device) not sent: \(error)")
}
}
/// Block until a token arrives or `timeout` elapses. Returns whether one did.
///
/// ⚠️ A `false` return means "nothing arrived", not "nothing is there" — the
/// caller must still run the cursor query, because a token can be missed
/// (Redis restarted, the cap trimmed it) and the database is the truth.
static func wait(device: String, timeout: Int, on app: Application) async -> Bool {
guard app.storage[EnabledKey.self] == true else { return false }
do {
let popped = try await app.redis
.brpop(from: key(device: device), timeout: .seconds(Int64(timeout)))
.get()
return popped != nil
} catch {
app.logger.debug("wake-up wait for \(device) failed: \(error)")
return false
}
}
struct EnabledKey: StorageKey { typealias Value = Bool }
}
Tests/FlamenetRelayTests/DeleteOnAckTests.swift
+142
View File
@@ -0,0 +1,142 @@
import Fluent
import FluentSQLiteDriver
import Foundation
import Testing
import Vapor
@testable import FlamenetRelay
/// The gate on `FN_DELETE_ON_ACK`.
///
/// 🔴 **This deletes ciphertext and there is no way to get it back.** The flag
/// stays off until these pass, because the first deploy could not tell a
/// working delete from a wiped test device — an empty thread proves nothing
/// either way. What matters is not "did something get deleted" but *which rows
/// went*, so every test here asserts on the survivors too.
@Suite("delete-on-ack")
struct DeleteOnAckTests {
/// A throwaway database with the real Envelope schema.
private func withDB(_ body: (any Database) async throws -> Void) async throws {
let app = try await Application.make(.testing)
defer { Task { try? await app.asyncShutdown() } }
app.databases.use(.sqlite(.memory), as: .sqlite)
app.migrations.add(CreateEnvelope())
// ⚠️ envelope_id arrives in a later migration, and the model expects
// it. Without this every insert fails with "no column named
// envelope_id" — a schema gap that reads like a test bug.
app.migrations.add(AddEnvelopeID())
try await app.autoMigrate()
try await body(app.db)
}
private func envelope(to device: String, on db: any Database) async throws -> Int {
let e = Envelope()
e.toUser = 1; e.toDevice = device
e.fromUser = 2; e.fromDevice = "sender-device"
e.payload = "ciphertext"; e.delivered = false
try await e.create(on: db)
return try #require(e.id)
}
private func ids(for device: String, on db: any Database) async throws -> [Int] {
try await Envelope.query(on: db)
.filter(\.$toDevice == device)
.sort(\.$id, .ascending)
.all()
.compactMap(\.id)
}
@Test("an acknowledged cursor deletes only at or below itself")
func deletesOnlyAcknowledged() async throws {
try await withDB { db in
let a = try await envelope(to: "device-A", on: db)
let b = try await envelope(to: "device-A", on: db)
let c = try await envelope(to: "device-A", on: db)
// The client has ingested up to `b`. `c` arrived after it asked.
try await Routes.deleteAcknowledged(toDevice: "device-A", after: b, on: db)
let left = try await ids(for: "device-A", on: db)
#expect(left == [c], "expected only the unacknowledged envelope to survive")
#expect(!left.contains(a))
#expect(!left.contains(b))
}
}
/// 🔴 **Why this feature is Postgres-only.**
///
/// The safety of delete-on-ack rests entirely on row ids never being
/// reused: the cursor is an id, and "delete everything at or below it"
/// is only sound if nothing below it can be new.
///
/// SQLite reuses rowids. Delete the highest row and the next insert takes
/// its id straight back — measured here, `first=1` then `late=1` — so a
/// message arriving after the cursor is deleted **unread**, with nothing
/// left to show it existed. Postgres sequences never reuse, which is why
/// the flag is refused on any other database rather than merely
/// discouraged.
///
/// This test asserts the hazard, not the fix. If it ever starts failing,
/// SQLite changed its allocation and the guard can be revisited.
@Test("SQLite reuses row ids, which is why delete-on-ack is Postgres-only")
func sqliteReusesRowIDs() async throws {
try await withDB { db in
let first = try await envelope(to: "device-A", on: db)
try await Routes.deleteAcknowledged(toDevice: "device-A", after: first, on: db)
let late = try await envelope(to: "device-A", on: db)
#expect(late == first,
"SQLite no longer reuses rowids — revisit the Postgres-only guard")
}
}
/// The same sequence on a database that does not reuse ids: the late
/// arrival keeps a higher id and survives the cursor.
@Test("a message that arrives after the cursor survives when ids are monotonic")
func laterArrivalSurvivesWithMonotonicIDs() async throws {
try await withDB { db in
let a = try await envelope(to: "device-A", on: db)
let b = try await envelope(to: "device-A", on: db) // keeps the high-water mark
try await Routes.deleteAcknowledged(toDevice: "device-A", after: a, on: db)
// `b` outlived the cursor, so the next id cannot collide with it.
let late = try await envelope(to: "device-A", on: db)
#expect(late > a)
try await Routes.deleteAcknowledged(toDevice: "device-A", after: a, on: db)
#expect(try await ids(for: "device-A", on: db) == [b, late],
"a message that arrived after the cursor was deleted unread")
}
}
/// 🔴 Two devices on one account have independent cursors. The slower
/// device's mail is not the faster one's to drop.
@Test("one device's cursor does not touch another's inbox")
func scopedToOneDevice() async throws {
try await withDB { db in
let mine = try await envelope(to: "device-A", on: db)
let theirs = try await envelope(to: "device-B", on: db)
// B's id is higher, so an unscoped delete would take it too.
try await Routes.deleteAcknowledged(toDevice: "device-A", after: theirs, on: db)
#expect(try await ids(for: "device-A", on: db).isEmpty)
#expect(try await ids(for: "device-B", on: db) == [theirs],
"another device's inbox was deleted")
_ = mine
}
}
/// 🔴 A cursor of zero means "I have ingested nothing" — a fresh install, a
/// reset, a client that lost its cursor. Read as a bound it would delete
/// every row for that device.
@Test("a zero cursor deletes nothing")
func zeroCursorIsNotABound() async throws {
try await withDB { db in
let a = try await envelope(to: "device-A", on: db)
let b = try await envelope(to: "device-A", on: db)
try await Routes.deleteAcknowledged(toDevice: "device-A", after: 0, on: db)
#expect(try await ids(for: "device-A", on: db) == [a, b],
"a reset cursor wiped the inbox")
}
}
}