datadog_agent_remote_config/subscription.rs
1//! Typed subscriptions to a product's configuration.
2
3use std::fmt;
4use std::sync::Arc;
5
6use tokio::sync::watch;
7
8use crate::decoder::Outcome;
9
10/// The published state of one product.
11///
12/// A rejection never evicts `accepted`, so retaining the last known-good configuration is the client's behavior rather
13/// than each consumer's bookkeeping.
14pub(crate) struct Snapshot<T, E> {
15 /// The most recently accepted configuration, absent until the first snapshot is accepted.
16 pub(crate) accepted: Option<Arc<T>>,
17
18 /// The rejection of the most recent snapshot, absent while the most recent snapshot was accepted.
19 pub(crate) rejection: Option<Arc<E>>,
20}
21
22/// A handle that observes one product's configuration as the client publishes it.
23///
24/// `T` is the snapshot the product's [`ProductDecoder`](crate::ProductDecoder) builds and `E` is the error it produces,
25/// so a subscription reads as the value it delivers rather than as the decoder that produced it. A product with nothing
26/// richer to report leaves `E` at its default of [`String`].
27///
28/// A consumer reads [`current`](Self::current) and then loops on [`changed`](Self::changed). A subscription created
29/// after the client has already published treats that value as observed and receives no notification for it, so a
30/// consumer that only awaited `changed` would wait for a snapshot that may never arrive.
31///
32/// Cloning shares one subscription between several consumers: each clone tracks its own position, while decoding
33/// happens once per snapshot. Slow consumers may skip intermediate publications and observe only the latest state.
34/// Dropping the last clone unsubscribes the product.
35pub struct Subscription<T, E = String> {
36 pub(crate) receiver: watch::Receiver<Snapshot<T, E>>,
37}
38
39impl<T, E> Subscription<T, E> {
40 /// Creates a subscription that never receives a snapshot.
41 ///
42 /// Use this where a component takes a subscription but the process has no client to subscribe with, for example
43 /// because Remote Configuration is turned off. The component then runs the same code path as one whose client has
44 /// not delivered anything yet: [`current`](Self::current) always returns `None`, and [`changed`](Self::changed)
45 /// waits indefinitely.
46 ///
47 /// An inert subscription is not registered with any client, so it does not count against a client's one
48 /// subscription per product.
49 pub fn inert() -> Self {
50 // `changed` turns a closed channel into a future that never completes.
51 let (publisher, subscription) = Publisher::new();
52 drop(publisher);
53 subscription
54 }
55
56 /// Returns the most recently accepted configuration.
57 ///
58 /// Returns `None` until the first snapshot is accepted, whether because the worker has not yet polled, is not
59 /// running, or the subscription is [inert](Self::inert). `None` does not mean the product has no configuration
60 /// assigned: a fresh decoder's [`build`](crate::ProductDecoder::build) must produce a snapshot even for an empty
61 /// assignment.
62 pub fn current(&self) -> Option<Arc<T>> {
63 self.receiver.borrow().accepted.clone()
64 }
65
66 /// Waits for a newly published snapshot.
67 ///
68 /// This takes `&mut self` because the subscription tracks which publication it has observed. Each task that waits
69 /// holds its own clone.
70 ///
71 /// Slow consumers may skip intermediate publications and observe only the latest state. Once the worker stops,
72 /// this waits indefinitely after any pending publication has been observed, so a caller may `select!` on it.
73 ///
74 /// A published snapshot is not guaranteed to differ from the previous one: after the worker restarts, it decodes
75 /// every product again. An accepted update can also retain the same `Arc` when the product's
76 /// [`keeps_previous`](crate::ProductDecoder::keeps_previous) returns `true`.
77 ///
78 /// # Errors
79 ///
80 /// Returns the subscriber's own decoding error when the snapshot was rejected. The client reports that rejection to
81 /// the Agent independently; a consumer with nothing to report can ignore it. [`current`](Self::current) continues
82 /// to return the last accepted configuration.
83 pub async fn changed(&mut self) -> Result<Arc<T>, Arc<E>> {
84 if self.receiver.changed().await.is_err() {
85 return std::future::pending().await;
86 }
87
88 let snapshot = self.receiver.borrow_and_update();
89 if let Some(error) = &snapshot.rejection {
90 Err(Arc::clone(error))
91 } else {
92 Ok(Arc::clone(
93 snapshot.accepted.as_ref().expect("published snapshot has a value"),
94 ))
95 }
96 }
97}
98
99impl<T, E> fmt::Debug for Subscription<T, E> {
100 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
101 let accepted = self.receiver.borrow().accepted.is_some();
102 f.debug_struct("Subscription")
103 .field("accepted", &accepted)
104 .finish_non_exhaustive()
105 }
106}
107
108impl<T, E> Clone for Subscription<T, E> {
109 fn clone(&self) -> Self {
110 Self {
111 receiver: self.receiver.clone(),
112 }
113 }
114}
115
116/// The publishing side of one product's [`Subscription`] and its clones.
117///
118/// The client's registry and [`TestPublisher`](crate::TestPublisher) both publish through this, so a subscription
119/// behaves the same in a subscriber's test as in production.
120pub(crate) struct Publisher<T, E> {
121 sender: watch::Sender<Snapshot<T, E>>,
122}
123
124impl<T, E> Publisher<T, E> {
125 /// Creates a publisher and a subscription with no accepted snapshot.
126 pub(crate) fn new() -> (Self, Subscription<T, E>) {
127 let (sender, receiver) = watch::channel(Snapshot {
128 accepted: None,
129 rejection: None,
130 });
131 (Self { sender }, Subscription { receiver })
132 }
133
134 pub(crate) fn accept(&self, snapshot: T) {
135 // The replaced values are dropped after `send_modify` releases the channel's lock, so a subscriber's `Drop` for
136 // its previous snapshot or rejection can read the subscription instead of deadlocking. A panic in that `Drop`
137 // escapes the decoder's `catch_unwind` and ends the worker, which its supervisor restarts.
138 let mut replaced = (None, None);
139 self.sender.send_modify(|state| {
140 replaced = (state.accepted.replace(Arc::new(snapshot)), state.rejection.take());
141 });
142 drop(replaced);
143 }
144
145 /// Clears any rejection and notifies subscribers, publishing `snapshot` only if none has been accepted yet.
146 pub(crate) fn keep_previous(&self, snapshot: T) {
147 // Dropped after the lock is released, as in `accept`.
148 let mut replaced = (None, None);
149 self.sender.send_modify(|state| {
150 let unused = match &state.accepted {
151 Some(_) => Some(snapshot),
152 None => {
153 state.accepted = Some(Arc::new(snapshot));
154 None
155 }
156 };
157 replaced = (unused, state.rejection.take());
158 });
159 drop(replaced);
160 }
161
162 pub(crate) fn reject(&self, error: E) {
163 // Dropped after the lock is released, as in `accept`.
164 let mut replaced = None;
165 self.sender.send_modify(|state| {
166 replaced = state.rejection.replace(Arc::new(error));
167 });
168 drop(replaced);
169 }
170
171 /// Publishes the outcome of a decoder run. A panicked run publishes nothing, so subscribers are not notified.
172 pub(crate) fn publish(&self, outcome: Outcome<T, E>) {
173 match outcome {
174 Outcome::Accepted(snapshot) => self.accept(snapshot),
175 Outcome::KeepsPrevious(snapshot) => self.keep_previous(snapshot),
176 Outcome::Rejected { error, .. } => self.reject(error),
177 Outcome::Panicked => {}
178 }
179 }
180
181 /// Returns whether any clone of the subscription is still alive.
182 ///
183 /// Once this is false it stays false: a subscription can only be cloned from a live one.
184 pub(crate) fn is_subscribed(&self) -> bool {
185 self.sender.receiver_count() > 0
186 }
187}