From 1eaca27804748d6504b8a8ac3db1efebb76decb1 Mon Sep 17 00:00:00 2001 From: "torlando-agent[bot]" <281092095+torlando-agent[bot]@users.noreply.github.com> Date: Tue, 28 Jul 2026 17:52:46 +0000 Subject: [PATCH] fix: serialize location share queue work --- .../Telemetry/LocationShareScheduler.cpp | 93 +++++++++++----- .../Telemetry/LocationShareScheduler.h | 26 ++++- .../native/test_location_share_scheduler.cpp | 105 ++++++++++++++---- 3 files changed, 176 insertions(+), 48 deletions(-) diff --git a/lib/tdeck_ui/Telemetry/LocationShareScheduler.cpp b/lib/tdeck_ui/Telemetry/LocationShareScheduler.cpp index 8d57e7ca..6ad6c994 100644 --- a/lib/tdeck_ui/Telemetry/LocationShareScheduler.cpp +++ b/lib/tdeck_ui/Telemetry/LocationShareScheduler.cpp @@ -116,6 +116,32 @@ uint64_t LocationShareScheduler::retryDelay(uint8_t failure_count) { return delay > MAX_RETRY_MILLIS ? MAX_RETRY_MILLIS : delay; } +void LocationShareScheduler::scheduleRejectedWork( + ShareSession& session, + ShareWorkType type, + uint64_t now_millis) { + session.awaiting_ack = false; + session.pending_token = 0; + session.ack_deadline_millis = 0; + if (type == ShareWorkType::LOCATION && + (session.cease_pending || + (session.has_expiry && now_millis >= session.expires_at_millis))) { + session.cease_pending = true; + session.failure_count = 0; + session.next_attempt_millis = now_millis; + return; + } + if (session.failure_count < std::numeric_limits::max()) { + ++session.failure_count; + } + uint64_t next = boundedAdd(now_millis, retryDelay(session.failure_count)); + if (type == ShareWorkType::LOCATION && session.has_expiry && + next > session.expires_at_millis) { + next = session.expires_at_millis; + } + session.next_attempt_millis = next; +} + std::size_t LocationShareScheduler::find(const PeerId& peer) const { for (std::size_t index = 0; index < MAX_SHARE_SESSIONS; ++index) { if (slots_[index].occupied && @@ -149,9 +175,13 @@ void LocationShareScheduler::clear(std::size_t index) { void LocationShareScheduler::observeClock(uint64_t now_millis) { if (last_observed_millis_ != 0 && now_millis < last_observed_millis_) { for (std::size_t index = 0; index < MAX_SHARE_SESSIONS; ++index) { - if (slots_[index].occupied && - !slots_[index].session.awaiting_ack) { - slots_[index].session.next_attempt_millis = now_millis; + if (!slots_[index].occupied) continue; + ShareSession& session = slots_[index].session; + if (session.awaiting_ack) { + session.ack_deadline_millis = + boundedAdd(now_millis, ACKNOWLEDGEMENT_LEASE_MILLIS); + } else { + session.next_attempt_millis = now_millis; } } } @@ -178,6 +208,9 @@ ShareSessionResult LocationShareScheduler::start( std::size_t target = find(peer); const bool updating = target != NO_SLOT; + if (updating && slots_[target].session.awaiting_ack) { + return ShareSessionResult::BUSY; + } if (!updating) { target = firstVacant(); if (target == NO_SLOT) return ShareSessionResult::CAPACITY; @@ -218,6 +251,9 @@ ShareSessionResult LocationShareScheduler::restore( observeClock(now_millis); std::size_t target = find(peer); + if (target != NO_SLOT && slots_[target].session.awaiting_ack) { + return ShareSessionResult::BUSY; + } if (target == NO_SLOT) { target = firstVacant(); if (target == NO_SLOT) return ShareSessionResult::CAPACITY; @@ -252,8 +288,9 @@ ShareSessionResult LocationShareScheduler::stop( observeClock(now_millis); ShareSession& session = slots_[index].session; session.cease_pending = true; - session.awaiting_ack = false; - session.pending_token = 0; + if (session.awaiting_ack) { + return ShareSessionResult::STOPPING; + } session.failure_count = 0; session.next_attempt_millis = now_millis; return ShareSessionResult::STOPPING; @@ -262,6 +299,7 @@ ShareSessionResult LocationShareScheduler::stop( bool LocationShareScheduler::cancelWithoutCease(const PeerId& peer) { const std::size_t index = find(peer); if (index == NO_SLOT) return false; + if (slots_[index].session.awaiting_ack) return false; clear(index); return true; } @@ -286,6 +324,11 @@ SharePollResult LocationShareScheduler::poll( session.failure_count = 0; if (!session.awaiting_ack) session.next_attempt_millis = now_millis; } + if (session.awaiting_ack && + now_millis >= session.ack_deadline_millis) { + const ShareWorkType expired_type = session.pending_type; + scheduleRejectedWork(session, expired_type, now_millis); + } if (session.awaiting_ack || now_millis < session.next_attempt_millis) { continue; } @@ -299,6 +342,8 @@ SharePollResult LocationShareScheduler::poll( candidate.peer = session.peer; candidate.type = type; candidate.token = nextToken(); + candidate.ack_deadline_millis = + boundedAdd(now_millis, ACKNOWLEDGEMENT_LEASE_MILLIS); candidate.has_expiry = session.has_expiry; candidate.expires_at_millis = session.expires_at_millis; candidate.approx_radius_meters = session.approx_radius_meters; @@ -306,6 +351,7 @@ SharePollResult LocationShareScheduler::poll( session.awaiting_ack = true; session.pending_type = type; session.pending_token = candidate.token; + session.ack_deadline_millis = candidate.ack_deadline_millis; output = candidate; return SharePollResult::WORK; } @@ -333,6 +379,7 @@ ShareAckResult LocationShareScheduler::acknowledge( const ShareWorkType acknowledged_type = session.pending_type; session.awaiting_ack = false; session.pending_token = 0; + session.ack_deadline_millis = 0; if (queue_accepted) { session.failure_count = 0; if (acknowledged_type == ShareWorkType::CEASE) { @@ -356,22 +403,7 @@ ShareAckResult LocationShareScheduler::acknowledge( return ShareAckResult::ACCEPTED; } - if (session.failure_count < std::numeric_limits::max()) { - ++session.failure_count; - } - if (acknowledged_type == ShareWorkType::LOCATION && session.has_expiry && - now_millis >= session.expires_at_millis) { - session.cease_pending = true; - session.failure_count = 0; - session.next_attempt_millis = now_millis; - } else { - uint64_t next = boundedAdd(now_millis, retryDelay(session.failure_count)); - if (acknowledged_type == ShareWorkType::LOCATION && session.has_expiry && - next > session.expires_at_millis) { - next = session.expires_at_millis; - } - session.next_attempt_millis = next; - } + scheduleRejectedWork(session, acknowledged_type, now_millis); return ShareAckResult::RETRY_SCHEDULED; } @@ -384,14 +416,18 @@ bool LocationShareScheduler::get( return true; } -std::size_t LocationShareScheduler::snapshot( +ShareSnapshotResult LocationShareScheduler::snapshot( ShareRestoreEntry* output, - std::size_t capacity) const { - if (output == nullptr || capacity == 0) return 0; + std::size_t capacity, + std::size_t& written_or_required) const { + written_or_required = size_; + if (capacity < size_) return ShareSnapshotResult::BUFFER_TOO_SMALL; + if (size_ != 0 && output == nullptr) { + return ShareSnapshotResult::INVALID_ARGUMENT; + } + std::size_t copied = 0; - for (std::size_t index = 0; - index < MAX_SHARE_SESSIONS && copied < capacity; - ++index) { + for (std::size_t index = 0; index < MAX_SHARE_SESSIONS; ++index) { if (!slots_[index].occupied) continue; const ShareSession& session = slots_[index].session; ShareRestoreEntry entry{}; @@ -405,7 +441,8 @@ std::size_t LocationShareScheduler::snapshot( entry.record.cease_pending = session.cease_pending; output[copied++] = entry; } - return copied; + written_or_required = copied; + return ShareSnapshotResult::OK; } } // namespace Telemetry diff --git a/lib/tdeck_ui/Telemetry/LocationShareScheduler.h b/lib/tdeck_ui/Telemetry/LocationShareScheduler.h index 9687e183..a2190894 100644 --- a/lib/tdeck_ui/Telemetry/LocationShareScheduler.h +++ b/lib/tdeck_ui/Telemetry/LocationShareScheduler.h @@ -13,6 +13,7 @@ constexpr uint32_t MIN_SHARE_CADENCE_MILLIS = 1000; constexpr uint32_t MAX_SHARE_CADENCE_MILLIS = 24U * 60U * 60U * 1000U; constexpr uint64_t INITIAL_RETRY_MILLIS = 5000; constexpr uint64_t MAX_RETRY_MILLIS = 5ULL * 60ULL * 1000ULL; +constexpr uint64_t ACKNOWLEDGEMENT_LEASE_MILLIS = 1000; enum class ShareDuration : uint8_t { MINUTES_15, @@ -32,6 +33,7 @@ enum class ShareSessionResult : uint8_t { CAPACITY, CLOCK_UNAVAILABLE, INVALID_ARGUMENT, + BUSY, }; enum class ShareWorkType : uint8_t { @@ -54,6 +56,12 @@ enum class ShareAckResult : uint8_t { CLOCK_UNAVAILABLE, }; +enum class ShareSnapshotResult : uint8_t { + OK, + BUFFER_TOO_SMALL, + INVALID_ARGUMENT, +}; + enum class MidnightResult : uint8_t { OK, CLOCK_UNAVAILABLE, @@ -96,6 +104,7 @@ struct ShareSession { bool awaiting_ack = false; ShareWorkType pending_type = ShareWorkType::LOCATION; uint64_t pending_token = 0; + uint64_t ack_deadline_millis = 0; uint8_t failure_count = 0; }; @@ -103,6 +112,7 @@ struct ShareWork { PeerId peer{}; ShareWorkType type = ShareWorkType::LOCATION; uint64_t token = 0; + uint64_t ack_deadline_millis = 0; bool has_expiry = false; uint64_t expires_at_millis = 0; int32_t approx_radius_meters = 0; @@ -132,6 +142,11 @@ public: ShareSessionResult stop(const PeerId& peer, uint64_t now_millis); bool cancelWithoutCease(const PeerId& peer); + // WORK is a short exclusive queue-attempt lease. The caller must attempt + // queueing synchronously, acknowledge before ack_deadline_millis, and must + // never enqueue the work after that deadline. stop() orders CEASE behind + // an in-flight LOCATION; start()/restore() report BUSY instead of + // invalidating externally borrowed work. SharePollResult poll( uint64_t now_millis, bool current_location_valid, @@ -144,7 +159,12 @@ public: uint64_t now_millis); bool get(const PeerId& peer, ShareSession& output) const; - std::size_t snapshot(ShareRestoreEntry* output, std::size_t capacity) const; + // Atomic: BUFFER_TOO_SMALL writes no entries and reports the required + // capacity in written_or_required. + ShareSnapshotResult snapshot( + ShareRestoreEntry* output, + std::size_t capacity, + std::size_t& written_or_required) const; std::size_t size() const { return size_; } private: @@ -157,6 +177,10 @@ private: static bool validCadence(uint32_t cadence_millis); static uint64_t boundedAdd(uint64_t value, uint64_t delta); static uint64_t retryDelay(uint8_t failure_count); + static void scheduleRejectedWork( + ShareSession& session, + ShareWorkType type, + uint64_t now_millis); std::size_t find(const PeerId& peer) const; std::size_t firstVacant() const; uint64_t nextToken(); diff --git a/tests/native/test_location_share_scheduler.cpp b/tests/native/test_location_share_scheduler.cpp index 084e7115..e59a3cc4 100644 --- a/tests/native/test_location_share_scheduler.cpp +++ b/tests/native/test_location_share_scheduler.cpp @@ -202,31 +202,84 @@ void expirationQueuesExactlyOneCeaseUntilAccepted() { true, work) == Telemetry::SharePollResult::NO_WORK); } -void stopAndUpdateInvalidateOutstandingWork() { +void unacknowledgedLocationLeaseCannotBlockExpiryForever() { + Telemetry::LocationShareScheduler scheduler; + constexpr uint64_t start = 1000; + CHECK(scheduler.start(peer(14), options(), start) == + Telemetry::ShareSessionResult::STARTED); + Telemetry::ShareSession session{}; + CHECK(scheduler.get(peer(14), session)); + const uint64_t expiry = session.expires_at_millis; + + Telemetry::ShareWork location_work{}; + CHECK(scheduler.poll(expiry - 1, true, location_work) == + Telemetry::SharePollResult::WORK); + CHECK(location_work.type == Telemetry::ShareWorkType::LOCATION); + CHECK(location_work.ack_deadline_millis > expiry); + CHECK(scheduler.poll(expiry, false, location_work) == + Telemetry::SharePollResult::NO_WORK); + + Telemetry::ShareWork cease_work{}; + CHECK(scheduler.poll(location_work.ack_deadline_millis, false, cease_work) == + Telemetry::SharePollResult::WORK); + CHECK(cease_work.type == Telemetry::ShareWorkType::CEASE); + CHECK(scheduler.acknowledge(peer(14), location_work.token, true, + location_work.ack_deadline_millis) == + Telemetry::ShareAckResult::STALE_TOKEN); + CHECK(scheduler.acknowledge(peer(14), cease_work.token, true, + location_work.ack_deadline_millis) == + Telemetry::ShareAckResult::CEASED); +} + +void repeatedStopPreservesTheSingleOutstandingCease() { + Telemetry::LocationShareScheduler scheduler; + CHECK(scheduler.start(peer(15), options(), 1000) == + Telemetry::ShareSessionResult::STARTED); + CHECK(scheduler.stop(peer(15), 1001) == + Telemetry::ShareSessionResult::STOPPING); + Telemetry::ShareWork work{}; + CHECK(scheduler.poll(1001, false, work) == Telemetry::SharePollResult::WORK); + CHECK(work.type == Telemetry::ShareWorkType::CEASE); + const uint64_t token = work.token; + CHECK(scheduler.stop(peer(15), 1002) == + Telemetry::ShareSessionResult::STOPPING); + Telemetry::ShareSession session{}; + CHECK(scheduler.get(peer(15), session)); + CHECK(session.awaiting_ack); + CHECK(session.pending_token == token); + CHECK(scheduler.poll(1002, false, work) == Telemetry::SharePollResult::NO_WORK); + CHECK(scheduler.acknowledge(peer(15), token, true, 1003) == + Telemetry::ShareAckResult::CEASED); +} + +void serializesUpdateAndStopBehindOutstandingWork() { Telemetry::LocationShareScheduler scheduler; constexpr uint64_t now = 10000; CHECK(scheduler.start(peer(4), options(), now) == Telemetry::ShareSessionResult::STARTED); Telemetry::ShareWork work{}; CHECK(scheduler.poll(now, true, work) == Telemetry::SharePollResult::WORK); - const uint64_t stale_token = work.token; + const uint64_t first_token = work.token; auto updated = options(Telemetry::ShareDuration::HOUR_1, 30000); CHECK(scheduler.start(peer(4), updated, now + 1) == + Telemetry::ShareSessionResult::BUSY); + CHECK(!scheduler.cancelWithoutCease(peer(4))); + CHECK(scheduler.acknowledge(peer(4), first_token, true, now + 2) == + Telemetry::ShareAckResult::ACCEPTED); + CHECK(scheduler.start(peer(4), updated, now + 3) == Telemetry::ShareSessionResult::UPDATED); - CHECK(scheduler.acknowledge(peer(4), stale_token, true, now + 2) == - Telemetry::ShareAckResult::STALE_TOKEN); - CHECK(scheduler.poll(now + 1, true, work) == Telemetry::SharePollResult::WORK); + CHECK(scheduler.poll(now + 3, true, work) == Telemetry::SharePollResult::WORK); - CHECK(scheduler.stop(peer(4), now + 2) == - Telemetry::ShareSessionResult::STOPPING); - CHECK(scheduler.acknowledge(peer(4), work.token, true, now + 3) == - Telemetry::ShareAckResult::STALE_TOKEN); - CHECK(scheduler.poll(now + 2, false, work) == Telemetry::SharePollResult::WORK); - CHECK(work.type == Telemetry::ShareWorkType::CEASE); - CHECK(scheduler.acknowledge(peer(4), work.token, true, now + 3) == - Telemetry::ShareAckResult::CEASED); CHECK(scheduler.stop(peer(4), now + 4) == + Telemetry::ShareSessionResult::STOPPING); + CHECK(scheduler.acknowledge(peer(4), work.token, true, now + 5) == + Telemetry::ShareAckResult::ACCEPTED); + CHECK(scheduler.poll(now + 5, false, work) == Telemetry::SharePollResult::WORK); + CHECK(work.type == Telemetry::ShareWorkType::CEASE); + CHECK(scheduler.acknowledge(peer(4), work.token, true, now + 6) == + Telemetry::ShareAckResult::CEASED); + CHECK(scheduler.stop(peer(4), now + 7) == Telemetry::ShareSessionResult::NOT_FOUND); } @@ -356,8 +409,18 @@ void snapshotsOnlyDurableConsentFieldsIntoCallerStorage() { CHECK(scheduler.start(peer(10), options(Telemetry::ShareDuration::INDEFINITE), 1000) == Telemetry::ShareSessionResult::STARTED); - Telemetry::ShareRestoreEntry entries[1]{}; - CHECK(scheduler.snapshot(entries, 1) == 1); + Telemetry::ShareRestoreEntry too_small[1]{}; + too_small[0].record.cadence_millis = 1234; + std::size_t written = 0; + CHECK(scheduler.snapshot(too_small, 1, written) == + Telemetry::ShareSnapshotResult::BUFFER_TOO_SMALL); + CHECK(written == 2); + CHECK(too_small[0].record.cadence_millis == 1234); + + Telemetry::ShareRestoreEntry entries[2]{}; + CHECK(scheduler.snapshot(entries, 2, written) == + Telemetry::ShareSnapshotResult::OK); + CHECK(written == 2); CHECK(entries[0].record.cadence_millis == 30000); CHECK(entries[0].record.approx_radius_meters == 25); CHECK(entries[0].record.has_expiry); @@ -366,7 +429,9 @@ void snapshotsOnlyDurableConsentFieldsIntoCallerStorage() { CHECK(scheduler.stop(peer(9), 1001) == Telemetry::ShareSessionResult::STOPPING); CHECK(entries[0].record.expires_at_millis == saved_expiry); - CHECK(scheduler.snapshot(entries, 1) == 1); + CHECK(scheduler.snapshot(entries, 2, written) == + Telemetry::ShareSnapshotResult::OK); + CHECK(written == 2); CHECK(entries[0].record.cease_pending); Telemetry::LocationShareScheduler restored; @@ -378,8 +443,8 @@ void snapshotsOnlyDurableConsentFieldsIntoCallerStorage() { Telemetry::SharePollResult::WORK); CHECK(work.type == Telemetry::ShareWorkType::CEASE); - CHECK(scheduler.snapshot(nullptr, 1) == 0); - CHECK(scheduler.snapshot(entries, 0) == 0); + CHECK(scheduler.snapshot(nullptr, 2, written) == + Telemetry::ShareSnapshotResult::INVALID_ARGUMENT); } void survivesDeterministicHundredThousandOperationStress() { @@ -440,7 +505,9 @@ int main() { requestsImmediateWorkAndAdvancesOnlyAfterAcceptance(); retriesFailuresWithBoundedBackoffWithoutExtendingExpiry(); expirationQueuesExactlyOneCeaseUntilAccepted(); - stopAndUpdateInvalidateOutstandingWork(); + unacknowledgedLocationLeaseCannotBlockExpiryForever(); + repeatedStopPreservesTheSingleOutstandingCease(); + serializesUpdateAndStopBehindOutstandingWork(); enforcesCapacityWithoutEvictingConsent(); restoresOnlyValidUnexpiredConsentAndRequiresGps(); handlesUnavailableAndBackwardClocksFailClosed();