mirror of
https://github.com/block/buzz.git
synced 2026-11-03 07:51:22 +00:00
fix(mobile/ios): retain exact active-second push history
Co-authored-by: npub1tquskdu6yc4h8l7xxtceculxw600grekeq0xg2ukqfrwl7vrzg3quz3gmp <58390b379a262b73ffc632f19c73e6769ef40f36c81e642b960246eff9831222@buzz.block.builderlab.xyz> Signed-off-by: npub1tquskdu6yc4h8l7xxtceculxw600grekeq0xg2ukqfrwl7vrzg3quz3gmp <58390b379a262b73ffc632f19c73e6769ef40f36c81e642b960246eff9831222@buzz.block.builderlab.xyz>
This commit is contained in:
parent
28f2f276a7
commit
a255c4e399
@@ -11,40 +11,64 @@ public struct PushCatchUpSelection: Sendable {
|
||||
}
|
||||
|
||||
public enum PushCatchUp {
|
||||
/// Build catch-up filters that preserve the lease's constraints while moving
|
||||
/// strictly beyond a composite cursor. A second, exact-id filter keeps the
|
||||
/// current cursor event available only as a duplicate-cleanup fallback.
|
||||
/// Build catch-up filters that preserve the lease's constraints. The broad
|
||||
/// timestamp filter finds newer events, while composite pages cover every gap
|
||||
/// in the active second's exact delivered-ID set. A separate exact-id filter
|
||||
/// keeps the last displayed event available only for duplicate cleanup.
|
||||
public static func queryFilters(
|
||||
subscriptions: [PushLeaseSubscription],
|
||||
cursor: PushEventPosition?,
|
||||
delivered: [PushEventPosition],
|
||||
lastDisplayed: PushEventPosition?,
|
||||
limit: Int
|
||||
) -> [[String: Any]] {
|
||||
subscriptions.flatMap { subscription in
|
||||
guard let cursor else {
|
||||
return [subscription.filter.queryFilter(since: nil, limit: limit)]
|
||||
}
|
||||
|
||||
var filters: [[String: Any]] = []
|
||||
if cursor.createdAt < Int.max {
|
||||
filters.append(
|
||||
subscription.filter.queryFilter(
|
||||
since: cursor.createdAt + 1,
|
||||
if let cursor {
|
||||
// `+ 1` is safe only as the strictly-later half of this pair. The
|
||||
// active-second filters below start at the bottom and drain same-second
|
||||
// rows without making event-ID ordering the dedupe authority.
|
||||
if cursor.createdAt < Int.max {
|
||||
filters.append(
|
||||
subscription.filter.queryFilter(
|
||||
since: cursor.createdAt + 1,
|
||||
limit: limit
|
||||
)
|
||||
)
|
||||
}
|
||||
|
||||
var activeHead = subscription.filter.queryFilter(
|
||||
since: cursor.createdAt,
|
||||
limit: limit
|
||||
)
|
||||
activeHead["until"] = cursor.createdAt
|
||||
filters.append(activeHead)
|
||||
|
||||
let activeDelivered = delivered
|
||||
.filter { $0.createdAt == cursor.createdAt }
|
||||
.sorted()
|
||||
for boundaryIndex in stride(
|
||||
from: limit - 1,
|
||||
to: activeDelivered.count,
|
||||
by: limit
|
||||
) {
|
||||
var page = subscription.filter.queryFilter(
|
||||
since: cursor.createdAt,
|
||||
limit: limit
|
||||
)
|
||||
)
|
||||
page["until"] = cursor.createdAt
|
||||
page["before_id"] = activeDelivered[boundaryIndex].id
|
||||
filters.append(page)
|
||||
}
|
||||
} else {
|
||||
filters.append(subscription.filter.queryFilter(since: nil, limit: limit))
|
||||
}
|
||||
|
||||
var sameSecond = subscription.filter.queryFilter(
|
||||
since: cursor.createdAt,
|
||||
limit: limit
|
||||
)
|
||||
sameSecond["until"] = cursor.createdAt
|
||||
sameSecond["before_id"] = cursor.id
|
||||
filters.append(sameSecond)
|
||||
|
||||
var duplicateFallback = subscription.filter.queryFilter(since: nil, limit: 1)
|
||||
duplicateFallback["ids"] = [cursor.id]
|
||||
filters.append(duplicateFallback)
|
||||
if let lastDisplayed {
|
||||
var duplicateFallback = subscription.filter.queryFilter(since: nil, limit: 1)
|
||||
duplicateFallback["ids"] = [lastDisplayed.id]
|
||||
filters.append(duplicateFallback)
|
||||
}
|
||||
return filters
|
||||
}
|
||||
}
|
||||
|
||||
@@ -27,34 +27,46 @@ public struct PushEventPosition: Codable, Equatable, Comparable, Sendable {
|
||||
public struct PushOriginState: Codable, Equatable, Sendable {
|
||||
public let cursor: PushEventPosition?
|
||||
public let delivered: [PushEventPosition]
|
||||
public let lastDisplayed: PushEventPosition?
|
||||
|
||||
public init(cursor: PushEventPosition? = nil, delivered: [PushEventPosition] = []) {
|
||||
public init(
|
||||
cursor: PushEventPosition? = nil,
|
||||
delivered: [PushEventPosition] = [],
|
||||
lastDisplayed: PushEventPosition? = nil
|
||||
) {
|
||||
self.cursor = cursor
|
||||
self.delivered = delivered
|
||||
self.lastDisplayed = lastDisplayed
|
||||
}
|
||||
|
||||
public func contains(eventID: String) -> Bool {
|
||||
delivered.contains { $0.id == eventID }
|
||||
lastDisplayed?.id == eventID || delivered.contains { $0.id == eventID }
|
||||
}
|
||||
|
||||
public func consuming(
|
||||
_ position: PushEventPosition,
|
||||
deliveredLimit: Int = PushConsumptionState.maximumDeliveredPerOrigin
|
||||
_ position: PushEventPosition
|
||||
) -> PushOriginState {
|
||||
let nextCursor = max(cursor ?? position, position)
|
||||
var byID = Dictionary(uniqueKeysWithValues: delivered.map { ($0.id, $0) })
|
||||
let movedToNewSecond = cursor.map { position.createdAt > $0.createdAt } ?? true
|
||||
var byID = Dictionary(
|
||||
uniqueKeysWithValues: delivered
|
||||
.filter { !movedToNewSecond || $0.createdAt == position.createdAt }
|
||||
.map { ($0.id, $0) }
|
||||
)
|
||||
byID[position.id] = position
|
||||
let bounded = byID.values
|
||||
let retained = byID.values
|
||||
.filter { $0.createdAt >= nextCursor.createdAt }
|
||||
.sorted()
|
||||
.suffix(deliveredLimit)
|
||||
return PushOriginState(cursor: nextCursor, delivered: Array(bounded))
|
||||
return PushOriginState(
|
||||
cursor: nextCursor,
|
||||
delivered: Array(retained),
|
||||
lastDisplayed: position
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
public struct PushConsumptionState: Codable, Equatable, Sendable {
|
||||
public static let version = 1
|
||||
public static let maximumDeliveredPerOrigin = 64
|
||||
public static let allowedFutureSkewSeconds = 300
|
||||
|
||||
public var origins: [String: PushOriginState]
|
||||
@@ -112,16 +124,15 @@ public struct PushConsumptionState: Codable, Equatable, Sendable {
|
||||
) -> Bool {
|
||||
let originState = state(for: origin)
|
||||
return position.createdAt <= now + allowedFutureSkew
|
||||
&& position >= (originState.cursor ?? position)
|
||||
&& position.createdAt >= (originState.cursor?.createdAt ?? position.createdAt)
|
||||
&& !originState.contains(eventID: position.id)
|
||||
}
|
||||
|
||||
public mutating func consume(
|
||||
_ position: PushEventPosition,
|
||||
for origin: String,
|
||||
deliveredLimit: Int = maximumDeliveredPerOrigin
|
||||
for origin: String
|
||||
) {
|
||||
origins[origin] = state(for: origin).consuming(position, deliveredLimit: deliveredLimit)
|
||||
origins[origin] = state(for: origin).consuming(position)
|
||||
}
|
||||
|
||||
public mutating func removeInactiveOrigins(_ activeOrigins: Set<String>) {
|
||||
|
||||
@@ -35,6 +35,36 @@ final class PushCatchUpTests: XCTestCase {
|
||||
XCTAssertEqual(second.map(\.wasPreviouslyConsumed), [false, true])
|
||||
}
|
||||
|
||||
func testLateLowerIDSiblingIsSelectedThenDuplicateFallbackStaysSeparate() {
|
||||
let subscription = self.subscription()
|
||||
let higherID = String(repeating: "f", count: 64)
|
||||
let lowerID = String(repeating: "0", count: 64)
|
||||
let high = event(id: higherID)
|
||||
let lateLow = event(id: lowerID)
|
||||
var state = PushConsumptionState()
|
||||
state.consume(PushEventPosition(createdAt: 1_000, id: higherID), for: "origin")
|
||||
|
||||
let originState = state.state(for: "origin")
|
||||
let filters = PushCatchUp.queryFilters(
|
||||
subscriptions: [subscription],
|
||||
cursor: originState.cursor,
|
||||
delivered: originState.delivered,
|
||||
lastDisplayed: originState.lastDisplayed,
|
||||
limit: 10
|
||||
)
|
||||
let relayPage = relayResponse(events: [lateLow, high], filters: filters)
|
||||
let selections = PushCatchUp.orderedSelections(
|
||||
events: relayPage,
|
||||
origin: "origin",
|
||||
subscriptions: [subscription],
|
||||
consumptionState: state,
|
||||
verify: { _ in true }
|
||||
)
|
||||
|
||||
XCTAssertEqual(selections.map(\.event.id), [lowerID, higherID])
|
||||
XCTAssertEqual(selections.map(\.wasPreviouslyConsumed), [false, true])
|
||||
}
|
||||
|
||||
func testCompositeSameSecondQueryReachesSuccessorsBeyondFirstRelayPage() {
|
||||
let subscription = self.subscription()
|
||||
let events = (0..<25).map { event(id: String(format: "%064x", $0)) }
|
||||
@@ -42,10 +72,12 @@ final class PushCatchUpTests: XCTestCase {
|
||||
var selected: [String] = []
|
||||
|
||||
while selected.count < events.count {
|
||||
let cursor = state.state(for: "origin").cursor
|
||||
let originState = state.state(for: "origin")
|
||||
let filters = PushCatchUp.queryFilters(
|
||||
subscriptions: [subscription],
|
||||
cursor: cursor,
|
||||
cursor: originState.cursor,
|
||||
delivered: originState.delivered,
|
||||
lastDisplayed: originState.lastDisplayed,
|
||||
limit: 10
|
||||
)
|
||||
let relayPage = relayResponse(events: events, filters: filters)
|
||||
|
||||
@@ -30,7 +30,7 @@ final class PushConsumptionStateTests: XCTestCase {
|
||||
XCTAssertNil(state.querySince(for: "loser-origin", now: 3_000))
|
||||
}
|
||||
|
||||
func testDeliveredEventIsNeverSelectedAgainAndHistoryIsBounded() {
|
||||
func testDeliveredEventIsNeverSelectedAgainAndActiveSecondHistoryIsExact() {
|
||||
var state = PushConsumptionState()
|
||||
for index in 0..<70 {
|
||||
state.consume(
|
||||
@@ -41,10 +41,10 @@ final class PushConsumptionStateTests: XCTestCase {
|
||||
let latest = PushEventPosition(createdAt: 1_000, id: String(format: "%064x", 69))
|
||||
|
||||
XCTAssertFalse(state.canSelect(latest, for: "origin", now: 2_000))
|
||||
XCTAssertEqual(state.state(for: "origin").delivered.count, 64)
|
||||
XCTAssertEqual(state.state(for: "origin").delivered.count, 70)
|
||||
}
|
||||
|
||||
func testEvictedDeliveredEventBelowCompositeCursorIsNeverSelectedAgain() {
|
||||
func testActiveSecondHistoryPrunesWhenTimestampAdvances() {
|
||||
var state = PushConsumptionState()
|
||||
for index in 0..<70 {
|
||||
state.consume(
|
||||
@@ -52,15 +52,24 @@ final class PushConsumptionStateTests: XCTestCase {
|
||||
for: "origin"
|
||||
)
|
||||
}
|
||||
let evicted = PushEventPosition(createdAt: 1_000, id: String(format: "%064x", 0))
|
||||
let sameSecondSuccessor = PushEventPosition(
|
||||
createdAt: 1_000,
|
||||
id: String(format: "%064x", 70)
|
||||
)
|
||||
let first = PushEventPosition(createdAt: 1_000, id: String(format: "%064x", 0))
|
||||
let nextSecond = PushEventPosition(createdAt: 1_001, id: "next")
|
||||
|
||||
XCTAssertFalse(state.hasConsumed(eventID: evicted.id, for: "origin"))
|
||||
XCTAssertFalse(state.canSelect(evicted, for: "origin", now: 2_000))
|
||||
XCTAssertTrue(state.canSelect(sameSecondSuccessor, for: "origin", now: 2_000))
|
||||
XCTAssertTrue(state.hasConsumed(eventID: first.id, for: "origin"))
|
||||
state.consume(nextSecond, for: "origin")
|
||||
|
||||
XCTAssertFalse(state.hasConsumed(eventID: first.id, for: "origin"))
|
||||
XCTAssertEqual(state.state(for: "origin").delivered, [nextSecond])
|
||||
}
|
||||
|
||||
func testLowerIDSiblingArrivingAfterHigherIDRemainsSelectable() {
|
||||
var state = PushConsumptionState()
|
||||
let higher = PushEventPosition(createdAt: 1_000, id: "ff")
|
||||
let laterLower = PushEventPosition(createdAt: 1_000, id: "00")
|
||||
|
||||
state.consume(higher, for: "origin")
|
||||
|
||||
XCTAssertTrue(state.canSelect(laterLower, for: "origin", now: 2_000))
|
||||
}
|
||||
|
||||
func testEventOlderThanCompositeCursorIsNotSelected() {
|
||||
|
||||
@@ -233,10 +233,12 @@ final class BuzzPushNotificationResolver: BuzzPushNotificationResolving {
|
||||
completion(.diagnostic("Open Buzz to refresh notification subscriptions."))
|
||||
return
|
||||
}
|
||||
let cursor = consumptionState.state(for: community.id).cursor
|
||||
let originState = consumptionState.state(for: community.id)
|
||||
let filters = PushCatchUp.queryFilters(
|
||||
subscriptions: subscriptions,
|
||||
cursor: cursor,
|
||||
cursor: originState.cursor,
|
||||
delivered: originState.delivered,
|
||||
lastDisplayed: originState.lastDisplayed,
|
||||
limit: 10
|
||||
)
|
||||
guard let body = try? JSONSerialization.data(withJSONObject: filters) else {
|
||||
|
||||
Reference in New Issue
Block a user