mirror of
https://github.com/torlando-tech/pyxis.git
synced 2026-08-22 02:29:51 +00:00
fix: serialize location share queue work
This commit is contained in:
torlando-agent[bot]
@@ -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<uint8_t>::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<uint8_t>::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
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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();
|
||||
|
||||
Reference in New Issue
Block a user