Repository navigation
deps: update dependency flagsmith/sdk-spec to v0.7.0 - #68
Open
renovate[bot] wants to merge 2 commits into
Open
renovate[bot] wants to merge 2 commits into
renovate[bot] wants to merge 2 commits into
Conversation
|
Important Review skippedBot user detected. To trigger a single review, invoke the
 |
khvn26
changed the title
renovate
Bot
changed the title
renovate
Bot
force-pushed
the
renovate/sdk-spec
branch
from
October 29, 2026 10:32
f88224f to
956146e
Compare
|
Spec v0.6.0 adds three
Unit and integration tests cover the event payload, exposure de-duplication, buffer overflow, flushing (timer, full buffer, explicit and on drop), retry and drop handling, and when an exposure is or isn't recorded. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🔇 Additional comments: 3
src/flagsmith/events.rs-270-274: 1
270-274:Minor · Potential issue · Heavy lift
⚠️ Unverified finding
We couldn't verify this from the code. It is shown for review, not as a verified issue.A transport error retains the entire POST batch without an event idempotency key visible in the payload, so a timeout after server acceptance may replay custom events.
A later flush can double-count the conversion.
This can occur when: The events API processes a conversion POST, but its response is lost or arrives after the client timeout.
sdk-spec/steps.rs-101-107: 1
101-107:Minor · Potential issue · Quick win
⚠️ Unverified finding
We couldn't verify this from the code. It is shown for review, not as a verified issue.The new spec event-producing steps return after enqueueing, without calling
flush_events; the experiment and explicit-exposure steps use the same asynchronous processor.Spec assertions intermittently observe no request even though the step returned success.
This can occur when: The v0.6 spec runner checks its mock events endpoint immediately after a step returns, before the default flush interval or client drop.
src/flagsmith/events.rs-70-75: 1
70-75:Minor · Potential issue · Quick win
⚠️ Unverified finding
We couldn't verify this from the code. It is shown for review, not as a verified issue.Exposure deduplication keys omit
metadata.experiment_id, althoughget_experiment_flagrecords that ID. Equal feature, identity, and variant values from different experiments can collapse into one event.The second experiment receives no exposure record.
This can occur when: An identity sees the same variant on a feature in two successive experiments before the buffer flushes.
-
-
- Source mode: Auto
- Repository: Unavailable
- Branch: Unavailable
- File: No configuration file from this repository applied
- Commit: Unavailable
- Layers: Organization UI
- Review profile: ASSERTIVE
- Plan: Advanced
- Run ID:
bda0ea3c-e0ed-43bb-a1b6-b5b7d462055b
-
.github/workflows/sdk-spec.ymlsdk-spec/Cargo.tomlsdk-spec/steps.rssrc/flagsmith/events.rssrc/flagsmith/mod.rssrc/flagsmith/models.rstests/integration_test.rs
Included review availability: This review used your included allowance. Your plan provides up to 1 included review per hour; 0 remain after this review.
-
| }; | ||
| let endpoint = format!("{events_api_url}{separator}{EVENTS_ENDPOINT}"); | ||
|
|
||
| let (tx, rx) = flume::unbounded(); |
There was a problem hiding this comment.
-
Script executed:
cd "$(git rev-parse --show-toplevel 2>/dev/null || pwd)" git rev-parse HEAD 2>/dev/null git cat-file -t 1d1a2bb9541ab9c143a8209f92c929a445acc8e9 2>/dev/null || echo "base-missing" nl -ba src/flagsmith/events.rs | sed -n '1,635p'
Repository: Flagsmith/flagsmith-rust-client
Length of output: 24888
Script executed:
rg -n -F -e 'EventBuffer' -e 'held' -e 'is_full' -e 'max_buffer_size' -e 'fn send' -e 'fn flush' -e 'flume::' -e 'Message::Track' -- src/ || test "$?" -eq 1
Repository: Flagsmith/flagsmith-rust-client
Length of output: 3220
Script executed:
rg -n -F -e 'track_exposure_event' -e 'track_event' -e 'events_timeout' -e 'events_flush_interval' -e 'EventProcessor' -e 'DEFAULT_EVENTS' -- src/ tests/ || test "$?" -eq 1 echo "--- base events.rs presence ---" git cat-file -e 1d1a2bb9541ab9c143a8209f92c929a445acc8e9:src/flagsmith/events.rs 2>/dev/null && echo present || echo absent-at-base git ls-tree -r --name-only 1d1a2bb9541ab9c143a8209f92c929a445acc8e9 -- src/flagsmith/ 2>/dev/null || echo "ls-tree-failed" echo "--- mod.rs event-related regions ---" nl -ba src/flagsmith/mod.rs | sed -n '40,160p'
Repository: Flagsmith/flagsmith-rust-client
Length of output: 10116
Script executed:
nl -ba src/flagsmith/mod.rs | sed -n '270,400p'
Repository: Flagsmith/flagsmith-rust-client
Length of output: 6444
Bound the event queue to the configured buffer size.
The channel at line 184 is unbounded. While the worker blocks in a slow successful flush, callers keep queuing events. If events arrive faster than max_buffer_size per flush, the queue grows until memory is exhausted. Bound the queue and drop events when it is full.
🐛 Suggested fix
-
- let (tx, rx) = flume::unbounded(); + let (tx, rx) = flume::bounded(max_buffer_size.max(1));
fn send(&self, event: Event) { - if self.tx.send(Message::Track(event)).is_err() { - warn!("Event processor has stopped; dropped an event"); - } + match self.tx.try_send(Message::Track(event)) { + Ok(()) => {} + Err(flume::TrySendError::Full(_)) => warn!("Event queue is full; dropped an event"), + Err(flume::TrySendError::Disconnected(_)) => { + warn!("Event processor has stopped; dropped an event") + } + } }
Comment on lines
+191
to
+192
| let message = | ||
| rx.recv_timeout(flush_interval.saturating_sub(last_flushed.elapsed())); |
There was a problem hiding this comment.
-
Script executed:
nl -ba src/flagsmith/events.rs | sed -n '1,140p'
Repository: Flagsmith/flagsmith-rust-client
Length of output: 5557
Script executed:
nl -ba src/flagsmith/events.rs | sed -n '140,330p'
Repository: Flagsmith/flagsmith-rust-client
Length of output: 7978
Script executed:
nl -ba src/flagsmith/events.rs | sed -n '330,470p'
Repository: Flagsmith/flagsmith-rust-client
Length of output: 5803
Script executed:
nl -ba src/flagsmith/events.rs | sed -n '470,635p'
Repository: Flagsmith/flagsmith-rust-client
Length of output: 6164
Script executed:
grep -n -A2 -E '^name = "flume"|^flume' Cargo.toml Cargo.lock || test "$?" -eq 1 echo '--- base presence ---' git cat-file -e 1d1a2bb9541ab9c143a8209f92c929a445acc8e9:src/flagsmith/events.rs 2>/dev/null && echo 'events.rs exists at base' || echo 'events.rs absent at base' git diff --stat 1d1a2bb9541ab9c143a8209f92c929a445acc8e9 f2963ccf0605a76047c5c29b285f393c92e409aa -- src/flagsmith/events.rs || test "$?" -eq 1 echo '--- base worker ---' git grep -n -E 'recv_timeout|held|is_full' 1d1a2bb9541ab9c143a8209f92c929a445acc8e9 -- src/ || test "$?" -eq 1
Repository: Flagsmith/flagsmith-rust-client
Length of output: 887
Web query:
flume 0.10.14 Receiver recv_timeout source: does it return a queued message when the timeout is zero or the deadline has passedResult:
<source_evidence> <source kind="find_in_page"> <title>android/vendor/flume-0.10.14/src/lib.rs - toolchain/sccache - Git at Google</title> <location>https://android.googlesource.com/toolchain/sccache/%2B/9d42d9fbb0e5c63a0273d2a9a40d19ff09a3c000/android/vendor/flume-0.10.14/src/lib.rs</location> <excerpt>citeturn2view1 [wordlim: 200] Crawled: today; Content type: text/html; Source: find({"ref_id":"turn1view0","pattern":"recv_sync(&self"}); Total lines: 1025 L512: | fn send_sync( L513: | &self, L514: | msg: T, L530: | wait_lock(&self.chan).sending L531: | .as_mut() L532: | .unwrap().1 L533: | .retain(|s| s.signal().as_ptr() != hook.signal().as_ptr()); L534: | } L535: | hook.try_take().map(|msg| if self.is_disconnected() { L536: | Err(TrySendTimeoutError::Disconnected(msg)) L537: | } else { L538: | Err(TrySendTimeoutError::Timeout(msg)) L539: | }) L540: | .unwrap_or(Ok(())) L541: | }) L542: | } else { L543: | hook.wait_send(&self.disconnected); L544: | L545: | match hook.try_take() { L546: | Some(msg) => Err(TrySendTimeoutError::Disconnected(msg)), L547: | None => Ok(()), L548: | } L549: | }, L550: | ) L551: | } L552: | L553: | fn recv<S: Signal, R: From<Result<T, TryRecvTimeoutError>>>( L554: | &self, L555: | should_block: bool, L556: | make_signal: impl FnOnce() -> Arc<Hook<T, S>>, L557: | do_block: impl FnOnce(Arc<Hook<T, S>>) -> R, L558: | ) -> R { L559: | let mut chan = wait_lock(&self.chan); L560: | chan.pull_pending(true); L561: | L562: | if let Some(msg) = chan.queue.pop_front() { L563: | drop(chan); L564: | Ok(msg).into() L565: | } else if self.is_disconnected() { L566: | drop(chan); L567: | Err(TryRecvTimeoutError::Disconnected).into() L568: | } else if should_block { L569: | let hook = make_signal(); L570: | chan.waiting.push_back(hook.clone()); L571: | drop(chan); L572: | L573: | do_block(hook) L574: | } else { L575: | drop(chan); L576: | Err(TryRecvTimeoutError::Empty).into() L577: | } L578: | } L579: | L580: | fn recv_sync(&self, block: Option<Option<Instant>>) -> Result<T, TryRecvTimeoutError> { L581: | self.recv( L582: | // should_block L583: | block.is_some(), L584: | // make_signal L585: | || Hook::slot(None, SyncSignal::default()), L586: | // do_block L587: | |hook| if let Some(deadline) = block.unwrap() { L588: | hook.wait_deadline_recv(&self.disconnected, deadline) L589: | .or_else(|timed_out| { L590: | if timed_out { // Remove our signal L591: | let hook: Arc<Hook<T, dyn Signal>> = hook.clone(); L592: | wait_lock(&self.chan).waiting L593: | .retain(|s| s.signal().as_ptr() != hook.signal().as_ptr()); L594: | } L595: | match hook.try_take() { L596: | Some(msg) => Ok(msg), L597: | None => { L598: | let disconnected = self.is_disconnected(); // Check disconnect *before* msg L599: | if let Some(msg) = wait_lock(&self.chan).queue.pop_front() { L600: | Ok(msg) L601: | } else if disconnected { L602: | Err(TryRecvTimeoutError::Disconnected) L603: | } else { L604: | Err(TryRecvTimeoutError::Timeout) L605: | } L606: | }, L607: | } L608: | }) L609: | } else { L610: | hook.wait_recv(&self.disconnected) L611: | .or_else(|| wait_lock(&self.chan).queue.pop_front()) L612: | .ok_or(TryRecvTimeoutError::Disconnected) L613: | }, L614: | ) L615: | } L616: | L617: | /// Disconnect anything listening on this channel (this will not prevent receivers receiving L618: | /// msgs that have already been sent) L619: | fn disconnect_all(&self) { L620: | self.disconnected.store(true, Ordering::Relaxed); L621: | L622: | let mut chan = wait_lock(&self.chan); L623: | chan.pull_pending(false); L624: | if let Some((_, sending)) = chan.sending.as_ref() { L625: | sending.iter().for_each(|hook| { L626: | hook.signal().fire(); L627: | }) L628: | } L629: | chan.waiting.iter().for_each(|hook| { L630: | hook.signal().fire(); L631: | }); L632: | } L633: | L634: | fn is_disconnected(&self) -> bool { L635: | self.disconnected.load(Ordering::SeqCst) L636: | } L637: | L638: | fn is_empty(&self) -> bool { L639: | self.len() == 0 L640: | } L641: | L642: | fn is_full(&self) -> bool { L643: | self.capacity().map(|cap| cap == self.len()).unwrap_or(false) L644: | } L645: | L646: | fn len(&self) -> usize { L647: | let mut chan = wait_lock(&self.chan); L648: | chan.pull_pending(false); L649: | chan.queue.len() L650: | } L651: | L652: | fn capacity(&self) -> Option<usize> { L653: | wait_lock(&self.chan).sending.as_ref().map(|(cap, _)| *cap) L654: | } L655: | } L656: | L657: | /// A transmitting end of a channel. L658: | pub struct Sender<T> { L659: | shared: Arc<Shared<T>>, L660: | } L661: | L662: | impl<T> Sender<T> { L663: | /// Attempt to send a value into the channel. If the channel is bounded and full, or all L664: | /// receivers have been dropped, an error is returned. If the channel associated with this L665: | /// sender is unbounded, this method has the same behaviour as [`Sender::send`]. L666: | pub fn try_send(&self, msg: T) -> Result<(), TrySendError<T>> { L667: | self.shared.send_sync(msg, None).map_err(|err| match err { L668: | TrySendTimeoutError::Full(msg) => TrySendError::Full(msg), L669: | TrySendTimeoutError::Disconnected(msg) => TrySendError::Disconnected(msg), L670: | _ => unreachable!(), L671: | }) L672: | } L673: | L674: | /// Send a value into the channel, returning an error if all receivers have been dropped. L675: | /// If the channel is bounded and is full, this method will block until space is available L676: | /// or all receivers have been dropped. If the channel is unbounded, this method will not L677: | /// block. L678: | pub fn send(&self, msg: T) -> Result<(), SendError<T>> { </excerpt> </source> <source kind="open_page"> <title>android/vendor/flume-0.10.14/src/lib.rs - toolchain/sccache - Git at Google</title> <location>https://android.googlesource.com/toolchain/sccache/%2B/9d42d9fbb0e5c63a0273d2a9a40d19ff09a3c000/android/vendor/flume-0.10.14/src/lib.rs</location> <excerpt>citeturn1view0 [wordlim: 200] Crawled: today; Content type: text/html; Source: open({"ref_id":"turn0search0","lineno":null}); Total lines: 1025 L16: | //! - **Flexible**: `Sender` and `Receiver` both implement `Send + Sync + Clone` L17: | //! - **Familiar**: Drop-in replacement for `std::sync::mpsc` L18: | //! - **Capable**: Additional features like MPMC support and send timeouts/deadlines L19: | //! - **Simple**: Few dependencies, minimal codebase, fast to compile L20: | //! - **Asynchronous**: `async` support, including mix 'n match with sync code L21: | //! - **Ergonomic**: Powerful `select`-like interface L22: | //! L23: | //! ## Example L24: | //! L25: | //! ``` L26: | //! let (tx, rx) = flume::unbounded(); L27: | //! L28: | //! tx.send(42).unwrap(); L29: | //! assert_eq!(rx.recv().unwrap(), 42); L30: | //! ``` L31: | L32: | #![deny(missing_docs)] L33: | L34: | #[cfg(feature = "select")] L35: | pub mod select; L36: | #[cfg(feature = "async")] L37: | pub mod r#async; L38: | L39: | mod signal; L40: | L41: | // Reexports L42: | #[cfg(feature = "select")] L43: | pub use select::Selector; L44: | L45: | use std::{ L46: | collections::VecDeque, L47: | sync::{Arc, atomic::{AtomicUsize, AtomicBool, Ordering}}, L48: | time::{Duration, Instant}, L49: | marker::PhantomData, L50: | thread, L51: | fmt, L52: | }; L53: | L54: | use spin1::{Mutex as Spinlock, MutexGuard as SpinlockGuard}; L55: | use crate::signal::{Signal, SyncSignal}; L56: | L57: | /// An error that may be emitted when attempting to send a value into a channel on a sender when L58: | /// all receivers are dropped. L59: | #[derive(Copy, Clone, PartialEq, Eq)] L60: | pub struct SendError<T>(pub T); L61: | L62: | impl<T> SendError<T> { L63: | /// Consume the error, yielding the message that failed to send. L64: | pub fn into_inner(self) -> T { self.0 } L65: | } L66: | L67: | impl<T> fmt::Debug for SendError<T> { L68: | fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { L69: | "SendError(..)".fmt(f) L70: | } L71: | } L72: | L73: | impl<T> fmt::Display for SendError<T> { L74: | fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { L75: | "sending on a closed channel".fmt(f) L76: | } L77: | } L78: | L79: | impl<T> std::error::Error for SendError<T> {} L80: | L81: | /// An error that may be emitted when attempting to send a value into a channel on a sender when L82: | /// the channel is full or all receivers are dropped. L83: | #[derive(Copy, Clone, PartialEq, Eq)] L84: | pub enum TrySendError<T> { L85: | /// The channel the message is sent on has a finite capacity and was full when the send was attempted. L86: | Full(T), L87: | /// All channel receivers were dropped and so the message has nobody to receive it. L88: | Disconnected(T), L89: | } L90: | L91: | impl<T> TrySendError<T> { L92: | /// Consume the error, yielding the message that failed to send. L93: | pub fn into_inner(self) -> T { L94: | match self { L95: | Self::Full(msg) | Self::Disconnected(msg) => msg, L96: | } L97: | } L98: | } L99: | L100: | impl<T> fmt::Debug for TrySendError<T> { L101: | fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { L102: | match *self { L103: | TrySendError::Full(..) => "Full(..)".fmt(f), L104: | TrySendError::Disconnected(..) => "Disconnected(..)".fmt(f), L105: | } L106: | } L107: | } L108: | L109: | impl<T> fmt::Display for TrySendError<T> { L110: | fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { L111: | match self { L112: | TrySendError::Full(..) => "sending on a full channel".fmt(f), L113: | TrySendError::Disconnected(..) => "sending on a closed channel".fmt(f), L114: | } L115: | } L116: | } L117: | L118: | impl<T> std::error::Error for TrySendError<T> {} L119: | L120: | impl<T> From<SendError<T>> for TrySendError<T> { L121: | fn from(err: SendError<T>) -> Self { L122: | match err { L123: | SendError(item) => Self::Disconnected(item), L124: | } L125: | } L126: | } L127: | L128: | /// An error that may be emitted when sending a value into a channel on a sender with a timeout when L129: | /// the send operation times out or all receivers are dropped. L130: | #[derive(Copy, Clone, PartialEq, Eq)] L131: | pub enum SendTimeoutError<T> { L132: | /// A timeout occurred when attempting to send the message. L133: | Timeout(T), L134: | /// All channel receivers were dropped and so the message has nobody to receive it. L135: | Disconnected(T), L136: | } L137: | L138: | impl<T> SendTimeoutError<T> { L139: | /// Consume the error, yielding the message that failed to send. L140: | pub fn into_inner(self) -> T { L141: | match self { L142: | Self::Timeout(msg) | Self::Disconnected(msg) => msg, L143: | } L144: | } L145: | } L146: | L147: | impl<T> fmt::Debug for SendTimeoutError<T> { L148: | fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { L149: | "SendTimeoutError(..)".fmt(f) L150: | } L151: | } L152: | L153: | impl<T> fmt::Display for SendTimeoutError<T> { L154: | fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { L155: | match self { L156: | SendTimeoutError::Timeout(..) => "timed out sending on a full channel".fmt(f), L157: | SendTimeoutError::Disconnected(..) => "sending on a closed channel".fmt(f), L158: | } L159: | } L160: | } L161: | L162: | impl<T> std::error::Error for SendTimeoutError<T> {} L163: | L164: | impl<T> From<SendError<T>> for SendTimeoutError<T> { L165: | fn from(err: SendError<T>) -> Self { L166: | match err { L167: | SendError(item) => Self::Disconnected(item), L168: | } L169: | } L170: | } L171: | L172: | enum TrySendTimeoutError<T> { L173: | Full(T), L174: | Disconnected(T), L175: | Timeout(T), L176: | } L177: | L178: | /// An error that may be emitted when attempting to wait for a value on a receiver when all senders L179: | /// are dropped and there are no more messages in the channel. L180: | #[derive(Copy, Clone, Debug, PartialEq, Eq)] L181: | pub enum RecvError { L182: | /// All senders were dropped and no messages are waiting in the channel, so no further messages can be received. L183: | Disconnected, L184: | } L185: | L186: | impl fmt::Display for RecvError { L187: | fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { L188: | match self { L189: | RecvError::Disconnected => "receiving on a closed channel".fmt(f), L190: | } L191: | } L192: | } L193: | L194: | impl std::error::Error for RecvError {} L195: | L196: | /// An error that may be emitted when attempting to fetch a value on a receiver when there are no L197: | /// messages in the channel. If there are no messages in the channel and all senders are dropped, L198: | /// then `TryRecvError::Disconnected` will be returned. L199: | #[derive(Copy, Clone, Debug, PartialEq, Eq)] L200: | pub enum TryRecvError { L201: | /// The channel was empty when the receive was attempted. L202: | Empty, L203: | /// All senders were dropped and no messages are waiting in the channel, so no further messages can be received. L204: | Disconnected, L205: | } L206: | L207: | impl fmt::Display for TryRecvError { L208: | fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { L209: | match self { L210: | TryRecvError::Empty => "receiving on an empty channel".fmt(f), L211: | TryRecvError::Disconnected => "channel is empty and closed".fmt(f), L212: | } L213: | } L214: | } L215: | L216: | impl std::error::Error for TryRecvError {} L217: | L218: | impl From<RecvError> for TryRecvError { L219: | fn from(err: RecvError) -> Self { L220: | match err { L221: | RecvError::Disconnected => Self::Disconnected, L222: | } L223: | } L224: | } L225: | L226: | /// An error that may be emitted when attempting to wait for a value on a receiver with a timeout L227: | /// when the receive operation times out or all senders are dropped and there are no values left L228: | /// in the channel. L229: | #[derive(Copy, Clone, Debug, PartialEq, Eq)] L230: | pub enum RecvTimeoutError { L231: | /// A timeout occurred when attempting to receive a message. L232: | Timeout, L233: | /// All senders were dropped and no messages are waiting in the channel, so no further messages can be received. L234: | Disconnected, L235: | } L236: | L237: | impl fmt::Display for RecvTimeoutError { L238: | fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { L239: | match self { L240: | RecvTimeoutError::Timeout => "timed out waiting on a channel".fmt(f), L241: | RecvTimeoutError::Disconnected => "channel is empty and closed".fmt(f), L242: | } L243: | } L244: | } L245: | L246: | impl std::error::Error for RecvTimeoutError {} L247: | L248: | impl From<RecvError> for RecvTimeoutError { L249: | fn from(err: RecvError) -> Self { L250: | match err { L251: | RecvError::Disconnected => Self::Disconnected, L252: | } L253: | } L254: | } L255: | L256: | enum TryRecvTimeoutError { L257: | Empty, L258: | Timeout, L259: | Disconnected, L260: | } L261: | L262: | // TODO: Investigate some sort of invalidation flag for timeouts L263: | struct Hook<T, S: ?Sized>(Option<Spinlock<Option<T>>>, S); L264: | L265: | impl<T, S: ?Sized + Signal> Hook<T, S> { L266: | pub fn slot(msg: Option<T>, signal: S) -> Arc<Self> where S: Sized { L267: | Arc::new(Self(Some(Spinlock::new(msg)), signal)) L268: | } L269: | L270: | pub fn trigger(signal: S) -> Arc<Self> where S: Sized { L271: | Arc::new(Self(None, signal)) L272: | } L273: | L274: | pub fn signal(&self) -> &S { L275: | &self.1 L276: | } L277: | L278: | pub fn fire_nothing(&self) -> bool { L279: | self.signal().fire() L280: | } L281: | L282: | pub fn fire_recv(&self) -> (T, &S) { L283: | let msg = self.0.as_ref().unwrap().lock().take().unwrap(); L284: | (msg, self.signal()) L285: | } L286: | L287: | pub fn fire_send(&self, msg: T) -> (Option<T>, &S) { L288: | let ret = match &self.0 { L289: | Some(hook) => { L290: | *hook.lock() = Some(msg); L291: | None L292: | }, L293: | None => Some(msg), L294: | }; L295: | (ret, self.signal()) L296: | } L297: | L298: | pub fn is_empty(&self) -> bool { L299: | self.0.as_ref().map(|s| s.lock().is_none()).unwrap_or(true) L300: | } L301: | L302: | pub fn try_take(&self) -> Option<T> { L303: | self.0.as_ref().and_then(|s| s.lock().take()) L304: | } L305: | } L306: | L307: | impl<T> Hook<T, SyncSignal> { L308: | pub fn wait_recv(&self, abort: &AtomicBool) -> Option<T> { L309: | loop { L310: | let disconnected = abort.load(Ordering::SeqCst); // Check disconnect *before* msg L311: | let msg = self.0.as_ref().unwrap().lock().take(); L312: | if let Some(msg) = msg { L313: | break Some(msg); L314: | } else if disconnected { L315: | break None; L316: | } else { L317: | self.signal().wait() L318: | } L319: | } L320: | } L321: | L322: | // Err(true) if timeout L323: | pub fn wait_deadline_recv(&self, abort: &AtomicBool, deadline: Instant) -> Result<T, bool> { L324: | loop { L325: | let disconnected = abort.load(Ordering::SeqCst); // Check disconnect *before* msg L326: | let msg = self.0.as_ref().unwrap().lock().take(); L327: | if let Some(msg) = msg { L328: | break Ok(msg); L329: | } else if disconnected { L330: | break Err(false); L331: | } else if let Some(dur) = deadline.checked_duration_since(Instant::now()) { L332: | self.signal().wait_timeout(dur); L333: | } else { L334: | break Err(true); L335: | } L336: | } L337: | } L338: | L339: | pub fn wait_send(&self, abort: &AtomicBool) { L340: | loop { L341: | let disconnected = abort.load(Ordering::SeqCst); // Check disconnect *before* msg L342: | if disconnected || self.0.as_ref().unwrap().lock().is_none() { L343: | break; L344: | } L345: | L346: | self.signal().wait(); L347: | } L348: | } L349: | L350: | // Err(true) if timeout L351: | pub fn wait_deadline_send(&self, abort: &AtomicBool, deadline: Instant) -> Result<(), bool> { L352: | loop { L353: | let disconnected = abort.load(Ordering::SeqCst); // Check disconnect *before* msg L354: | if self.0.as_ref().unwrap().lock().is_none() { L355: | break Ok(()); L356: | } else if disconnected { L357: | break Err(false); L358: | } else if let Some(dur) = deadline.checked_duration_since(Instant::now()) { L359: | self.signal().wait_timeout(dur); L360: | } else { L361: | break Err(true); L362: | } L363: | } L364: | } L365: | } L366: | L367: | #[cfg(feature = "spin")] L368: | #[inline] L369: | fn wait_lock<T>(lock: &Spinlock<T>) -> SpinlockGuard<T> { L370: | let mut i = 4; L371: | loop { L372: | for _ in 0..10 { L373: | if let Some(guard) = lock.try_lock() { L374: | return guard; L375: | } L376: | thread::yield_now(); L377: | } L378: | // Sleep for at most ~1 ms L379: | thread::sleep(Duration::from_nanos(1 << i.min(20))); L380: | i += 1; L381: | } L382: | } L383: | L384: | #[cfg(not(feature = "spin"))] L385: | #[inline] L386: | fn wait_lock<'a, T>(lock: &'a Mutex<T>) -> MutexGuard<'a, T> { L387: | lock.lock().unwrap() L388: | } L389: | L390: | #[cfg(not(feature = "spin"))] L391: | use std::sync::{Mutex, MutexGuard}; L392: | L669: | TrySendTimeoutError::Disconnected(msg) => TrySendError::Disconnected(msg), L670: | _ => unreachable!(), L671: | }) L672: | } L673: | L674: | /// Send a value into the channel, returning an error if all receivers have been dropped. L675: | /// If the channel is bounded and is full, this method will block until space is available L676: | /// or all receivers have been dropped. If the channel is unbounded, this method will not L677: | /// block. L678: | pub fn send(&self, msg: T) -> Result<(), SendError<T>> { L679: | self.shared.send_sync(msg, Some(None)).map_err(|err| match err { L680: | TrySendTimeoutError::Disconnected(msg) => SendError(msg), L681: | _ => unreachable!(), L682: | }) L683: | } L684: | L685: | /// Send a value into the channel, returning an error if all receivers have been dropped L686: | /// or the deadline has passed. If the channel is bounded and is full, this method will L687: | /// block until space is available, the deadline is reached, or all receivers have been L688: | /// dropped. L689: | pub fn send_deadline(&self, msg: T, deadline: Instant) -> Result<(), SendTimeoutError<T>> { L690: | self.shared.send_sync(msg, Some(Some(deadline))).map_err(|err| match err { L691: | TrySendTimeoutError::Disconnected(msg) => SendTimeoutError::Disconnected(msg), L692: | TrySendTimeoutError::Timeout(msg) => SendTimeoutError::Timeout(msg), L693: | _ => unreachable!(), L694: | }) L695: | } L696: | L697: | /// Send a value into the channel, returning an error if all receivers have been dropped L742: | impl<T> fmt::Debug for Sender<T> { L743: | fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { L744: | f.debug_struct("Sender").finish() L745: | } L746: | } L747: | L748: | impl<T> Drop for Sender<T> { L749: | fn drop(&mut self) { L750: | // Notify receivers that all senders have been dropped if the number of senders drops to 0. L751: | if self.shared.sender_count.fetch_sub(1, Ordering::Relaxed) == 1 { L752: | self.shared.disconnect_all(); L753: | } L754: | } L755: | } L756: | L757: | /// The receiving end of a channel. L758: | /// L759: | /// Note: Cloning the receiver *does not* turn this channel into a broadcast channel. L760: | /// Each message will only be received by a single receiver. This is useful for L761: | /// implementing work stealing for concurrent programs. L762: | pub struct Receiver<T> { L763: | shared: Arc<Shared<T>>, L764: | } L765: | L766: | impl<T> Receiver<T> { L767: | /// Attempt to fetch an incoming value from the channel associated with this receiver, L768: | /// returning an error if the channel is empty or if all senders have been dropped. L769: | pub fn try_recv(&self) -> Result<T, TryRecvError> { L770: | self.shared.recv_sync(None).map_err(|err| match err { L771: | TryRecvTimeoutError::Disconnected => TryRecvError::Disconnected, L772: | TryRecvTimeoutError::Empty => TryRecvError::Empty, L773: | _ => unreachable!(), L774: | }) L775: | } L77…[truncated]</excerpt> </source> <source kind="search"> <title>android/vendor/flume-0.10.14/src/lib.rs - toolchain/sccache - Git at Google</title> <location>https://android.googlesource.com/toolchain/sccache/%2B/9d42d9fbb0e5c63a0273d2a9a40d19ff09a3c000/android/vendor/flume-0.10.14/src/lib.rs</location> <excerpt>citeturn0search0 [wordlim: 200] Crawled: last week; | /// An error that may be emitted when attempting to wait for a value on a receiver with a timeout ... | RecvTimeoutError::Timeout => "timed out waiting on a channel".fmt(f), | } | } | } | | impl std::error::Error for TryRecvError {} | | impl From<RecvError> for TryRecvError { | fn from(err: RecvError) -> Self { | match err { | RecvError::Disconnected => Self::Disconnected, | } | } | } | | /// An error that may be emitted when attempting to wait for a value on a receiver with a timeout | /// when the receive operation times out or all senders are dropped and there are no values left | /// in the channel. | #[derive(Copy, Clone, Debug, PartialEq, Eq)] | pub enum RecvTimeoutError { | /// A timeout occurred when attempting to receive a message. | Timeout, | /// All senders were dropped and no messages are waiting in the channel, so no further messages can be received. | Disconnected, | } | | impl fmt::Display for RecvTimeoutError { | fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { | match self { | RecvTimeoutError::Timeout => "timed out waiting on a channel".fmt(f), | RecvTimeoutError::Disconnected => "channel is empty and closed".fmt(f), | TrySendTimeoutError::Disconnected(msg) => SendTimeoutError::Disconnected(msg), | TrySendTimeoutError::Timeout(msg) => SendTimeoutError::Timeout(msg), | _ => unreachable!(), | }) | } | | /// Send a value into the channel, returning an error if all receivers have been dropped | /// or the timeout has expired. If the channel is bounded and is full, this method will | /// block until space is available, the timeout has expired, or all receivers have been | /// dropped. | pub fn send_timeout(&self, msg: T, dur: Duration) -> Result<(), SendTimeoutError<T>> { | self.send_deadline(msg, Instant::now().checked_add(dur).unwrap()) | } | | /// Returns true if all receivers for this channel have been dropped. | pub fn is_disconnected(&self) -> bool { | self.shared.is_disconnected() | } | | /// Returns true if the channel is empty. | /// Note: Zero-capacity channels are always empty. | pub fn is_empty(&self) -> bool { | self.shared.is_empty() | } | | /// Returns true if the channel is full. | /// Note: Zero-capacity channels are always full. | pub fn is_full(&self) -> bool { | self.shared.is_full() | } | | /// Returns the number of messages in the channel | pub fn len(&self) -> usize { | /// Each message will only be received by a single receiver. This is useful for | /// implementing work stealing for concurrent programs. | pub struct Receiver<T> { | shared: Arc<Shared<T>>, | } | | impl<T> Receiver<T> { | /// Attempt to fetch an incoming value from the channel associated with this receiver, | /// returning an error if the channel is empty or if all senders have been dropped. | pub fn try_recv(&self) -> Result<T, TryRecvError> { | self.shared.recv_sync(None).map_err(|err| match err { | TryRecvTimeoutError::Disconnected => TryRecvError::Disconnected, | TryRecvTimeoutError::Empty => TryRecvError::Empty, | _ => unreachable!(), | }) | } | | /// Wait for an incoming value from the channel associated with this receiver, returning an | /// error if all senders have been dropped. | pub fn recv(&self) -> Result<T, RecvError> { | self.shared.recv_sync(Some(None)).map_err(|err| match err { | TryRecvTimeoutError::Disconnected => RecvError::Disconnected, | _ => unreachable!(), | }) | } | | /// Wait for an incoming value from the channel associated with this receiver, returning an | /// error if all senders have been dropped or the deadline has passed. | pub fn recv_deadline(&self, deadline: Instant) -> Result<T, RecvTimeoutError> { | self.shared.recv_sync(Some(Some(deadline))).map_err(|err| match err { | TryRecvTimeoutError::Disconnected => RecvTimeoutError::Disconnected, | TryRecvTimeoutError::Timeout => RecvTimeoutError::Timeout, | _ => unreachable!(), | }) | } | | /// Wait for an incoming value from the channel associated with this receiver, returning an | /// error if all senders have been dropped or the timeout has expired. | pub fn recv_timeout(&self, dur: Duration) -> Result<T, RecvTimeoutError> { | self.recv_deadline(Instant::now().checked_add(dur).unwrap()) | } | | /// Create a blocking iterator over the values received on the channel that finishes iteration | /// when all senders have been dropped. | /// | /// You can also create a self-owned iterator with [`Receiver::into_iter`]. | pub fn iter(&self) -> Iter<T> { | Iter { receiver: &self } | } | | /// A non-blocking iterator over the values received on the channel that finishes iteration </excerpt> </source> <source kind="search"> <title>android/vendor/flume-0.10.14/src/async.rs - toolchain/sccache - Git at Google</title> <location>https://android.googlesource.com/toolchain/sccache/%2B/9d42d9fbb0e5c63a0273d2a9a40d19ff09a3c000/android/vendor/flume-0.10.14/src/async.rs</location> <excerpt>citeturn0search1 [wordlim: 200] Crawled: 3 months ago; | RecvStream(RecvFut::new(self.0.receiver.clone())) </excerpt> </source> <source kind="search"> <title>lib.rs - source</title> <location>https://docs.rs/flume/latest/src/flume/lib.rs.html</location> <excerpt>citeturn0search2 [wordlim: 200] Crawled: 4 weeks ago; 229/// An error that may be emitted when attempting to wait for a value on a receiver with a timeout ... 243 RecvTimeoutError::Timeout => "timed out waiting on a channel".fmt(f), </excerpt> </source> <omitted_sources count="16">Additional source records were omitted from this bounded result.</omitted_sources> </source_evidence> Citations: - 1: https://android.googlesource.com/toolchain/sccache/%2B/9d42d9fbb0e5c63a0273d2a9a40d19ff09a3c000/android/vendor/flume-0.10.14/src/lib.rs - 2: https://android.googlesource.com/toolchain/sccache/%2B/9d42d9fbb0e5c63a0273d2a9a40d19ff09a3c000/android/vendor/flume-0.10.14/src/lib.rs
Script executed:
rg -n -F -e '.flush()' -e 'EventProcessor' -e 'track_exposure_event' -e 'track_event' -- src/ || test "$?" -eq 1
Repository: Flagsmith/flagsmith-rust-client
Length of output: 2896
Retry held events on the elapsed interval, not only on Timeout.
A failed send sets held, and the timed flush runs only on Timeout. flume returns queued messages before reporting a timeout, so a continuous backlog suppresses it. Held events stay unsent and can be evicted, even after the API recovers. Also check the interval in the Track arm.
🐛 Suggested fix
-
Ok(Message::Track(event)) => { buffer.push(event); - if buffer.is_full() && !buffer.held { - flush(&client, &endpoint, &mut buffer); - } + if (buffer.is_full() && !buffer.held) + || last_flushed.elapsed() >= flush_interval + { + flush(&client, &endpoint, &mut buffer); + last_flushed = Instant::now(); + } }
Comment on lines
+293
to
+295
| let event_processor = self.require_event_processor("use experiment flags")?; | ||
| let flag = self | ||
| .get_identity_flags(identifier, traits.clone(), None)? |
There was a problem hiding this comment.
-
Script executed:
sed -n '250,390p' src/flagsmith/mod.rs sed -n '910,1095p' tests/integration_test.rs rg -n 'identifier|experiment' sdk-spec
Repository: Flagsmith/flagsmith-rust-client
Length of output: 13099
Script executed:
set -u base=1d1a2bb9541ab9c143a8209f92c929a445acc8e9 git rev-parse --verify "${base}^{commit}" || echo "BASE_NOT_AVAILABLE" echo "--- base mod.rs exposure symbols ---" git show "${base}:src/flagsmith/mod.rs" 2>/dev/null | rg -n -F -e 'get_experiment_flag' -e 'track_exposure_event' -e 'require_event_processor' -e 'enable_events' || test "$?" -eq 1 echo "--- base events module presence ---" git ls-tree -r --name-only "${base}" 2>/dev/null | rg -n 'src/flagsmith/events' || test "$?" -eq 1
Repository: Flagsmith/flagsmith-rust-client
Length of output: 283
Skip exposure recording when the identifier is blank in get_experiment_flag.
get_experiment_flag records an exposure with identifier: Some(identifier.to_string()) and does not reject a blank identifier. The public track_exposure_event rejects a blank identifier, but get_experiment_flag calls the event processor directly and does not use that method. When a blank identifier returns an enrolled, enabled flag with a variant, the method enqueues an exposure with an empty identifier. The SDK's own exposure method refuses that input.
Suggested fix
-
let exposure = match (&flag.experiment, &flag.variant) { + _ if identifier.trim().is_empty() => Err("the identifier is blank"), _ if flag.is_default => Err("the flag is a default flag"),
renovate
Bot
changed the title
renovate
Bot
force-pushed
the
renovate/sdk-spec
branch
from
October 29, 2026 19:54
f2963cc to
bb61b96
Compare
Implements the experimentation features SDK spec v0.6.0 requires: - `enable_events` starts an event processor that buffers events and sends them to Flagsmith's events API in batches. - `get_experiment_flag` resolves a flag for an identity and records a `$flag_exposure` when the identity is enrolled in a running experiment. - `track_event` and `track_exposure_event` record events directly. - `Flag` exposes the `variant` and `experiment` served by remote identity evaluation. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> (cherry picked from commit f2963cc)
Author
Edited/Blocked NotificationRenovate will not automatically rebase this PR, because it does not recognize the last commit author and assumes somebody else may have edited the PR. You can manually request rebase by checking the rebase/retry box above. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
This PR contains the following updates:
v0.3.0→v0.7.0v0.3.0→v0.7.0Release Notes
Flagsmith/sdk-spec
v0.7.0Compare Source
What's Changed
New Contributors
Full Changelog: Flagsmith/sdk-spec@v0.6.0...v0.7.0
v0.6.0Compare Source
What's Changed
New Contributors
Full Changelog: Flagsmith/sdk-spec@v0.5.1...v0.6.0
v0.5.1Compare Source
What's Changed
Full Changelog: Flagsmith/sdk-spec@v0.5.0...v0.5.1
v0.5.0Compare Source
What's Changed
Full Changelog: Flagsmith/sdk-spec@v0.4.0...v0.5.0
v0.4.0Compare Source
What's Changed
New Contributors
Full Changelog: Flagsmith/sdk-spec@v0.3.0...v0.4.0
Configuration
📅 Schedule: (UTC)
🚦 Automerge: Disabled by config. Please merge this manually once you are satisfied.
♻ Rebasing: Whenever PR becomes conflicted, or you tick the rebase/retry checkbox.
🔕 Ignore: Close this PR and you won't be reminded about these updates again.
This PR was generated by Mend Renovate. View the repository job log.