risingwave_storage/hummock/event_handler/
version_update.rs

1// Copyright 2026 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::collections::VecDeque;
16use std::future::poll_fn;
17use std::task::Poll;
18
19use futures::FutureExt;
20use futures::future::BoxFuture;
21use risingwave_hummock_sdk::HummockVersionId;
22use risingwave_hummock_sdk::version::{HummockVersion, HummockVersionDelta};
23
24use super::HummockVersionUpdate;
25use crate::hummock::local_version::pinned_version::PinnedVersion;
26use crate::monitor::GLOBAL_CACHE_REFILL_METRICS;
27
28/// Completion permits this event to be published after earlier events.
29/// Refillers start their work independently of polling this future and define whether
30/// dropping it cancels or detaches the underlying work. It may be first-polled late.
31pub(super) type VersionUpdateCondition = BoxFuture<'static, ()>;
32
33/// One received payload's version transition, possibly containing multiple raw deltas.
34pub(super) struct VersionUpdateEvent {
35    /// Base ID for chain checks; published holders or the preceding event keep it pinned.
36    pub prev_version_id: HummockVersionId,
37    /// Constructed target awaiting publication.
38    pub new_pinned_version: PinnedVersion,
39}
40
41struct PendingVersion {
42    condition: VersionUpdateCondition,
43    event: VersionUpdateEvent,
44}
45
46/// Constructs received versions and orders their publication after cache refill completes.
47/// The refiller starts the work; this queue only waits for its completion.
48pub(super) struct VersionUpdateQueue {
49    /// Advanced by enqueue; may lead the version currently published to readers.
50    latest_received: PinnedVersion,
51    queue: VecDeque<PendingVersion>,
52}
53
54impl VersionUpdateQueue {
55    pub fn new(version: PinnedVersion) -> Self {
56        Self {
57            latest_received: version,
58            queue: VecDeque::new(),
59        }
60    }
61
62    /// Builds and pins the target version without advancing `latest_received` or publishing it.
63    /// Creating the pin sends a request to the pinned-version manager, even though this takes `&self`.
64    /// `before_apply` observes each intermediate version immediately before SDK delta application.
65    /// Returns `None` for an unchanged version, including an empty delta payload.
66    /// Enqueue the returned event before resolving another payload so its base is up to date.
67    ///
68    /// # Panics
69    /// Panics if the delta chain does not extend `latest_received`, or a snapshot is older than it.
70    pub fn resolve_update(
71        &self,
72        payload: HummockVersionUpdate,
73        mut before_apply: impl FnMut(&HummockVersion, &HummockVersionDelta),
74    ) -> Option<VersionUpdateEvent> {
75        let pinned_version = &self.latest_received;
76        let new_pinned_version = match payload {
77            HummockVersionUpdate::VersionDeltas(deltas) => {
78                let mut version = (**pinned_version).clone();
79                for delta in &deltas {
80                    assert_eq!(version.id, delta.prev_id);
81                    before_apply(&version, delta);
82                    version.apply_version_delta(delta);
83                }
84                pinned_version.new_with_local_version(version)?
85            }
86            HummockVersionUpdate::PinnedVersion(version) => {
87                pinned_version.new_pin_version(*version)?
88            }
89        };
90        Some(VersionUpdateEvent {
91            prev_version_id: pinned_version.id(),
92            new_pinned_version,
93        })
94    }
95
96    /// Advances `latest_received` and queues the transition with its publication condition.
97    /// Work startup and cancellation follow the [`VersionUpdateCondition`] contract.
98    ///
99    /// # Panics
100    /// Panics if the event's previous version does not match `latest_received`.
101    pub fn enqueue(&mut self, event: VersionUpdateEvent, condition: VersionUpdateCondition) {
102        assert_eq!(event.prev_version_id, self.latest_received.id());
103        self.latest_received = event.new_pinned_version.clone();
104        self.queue.push_back(PendingVersion { condition, event });
105        GLOBAL_CACHE_REFILL_METRICS.refill_queue_total.inc();
106    }
107
108    /// Removes up to 16 ready events from the front for the caller to publish in order.
109    /// Cancellation while pending preserves queued events and their condition futures.
110    /// An empty queue has no wake source: the caller must enqueue and repoll on incoming updates.
111    pub async fn next_events(&mut self) -> Vec<VersionUpdateEvent> {
112        poll_fn(|cx| {
113            const MAX_BATCH_SIZE: usize = 16;
114            let mut events = None;
115            while let Some(pending) = self.queue.front_mut()
116                && pending.condition.poll_unpin(cx).is_ready()
117            {
118                let pending = self.queue.pop_front().unwrap();
119                GLOBAL_CACHE_REFILL_METRICS.refill_queue_total.dec();
120                let events = events.get_or_insert_with(|| Vec::with_capacity(MAX_BATCH_SIZE));
121                events.push(pending.event);
122                if events.len() >= MAX_BATCH_SIZE {
123                    break;
124                }
125            }
126            match events {
127                Some(events) => Poll::Ready(events),
128                None => Poll::Pending,
129            }
130        })
131        .await
132    }
133}
134
135impl Drop for VersionUpdateQueue {
136    fn drop(&mut self) {
137        GLOBAL_CACHE_REFILL_METRICS
138            .refill_queue_total
139            .sub(self.queue.len() as i64);
140    }
141}
142
143#[cfg(test)]
144mod tests {
145    use tokio::sync::mpsc::unbounded_channel;
146
147    use super::*;
148
149    #[tokio::test]
150    async fn test_ready_prefix_limit_and_delta_observation() {
151        let version = PinnedVersion::new(HummockVersion::default(), unbounded_channel().0);
152        let mut queue = VersionUpdateQueue::new(version);
153        for id in 1..=17 {
154            let deltas = (id * 2 - 1..=id * 2)
155                .map(|raw_id| {
156                    let mut delta = HummockVersionDelta::default();
157                    delta.prev_id = (raw_id - 1).into();
158                    delta.id = raw_id.into();
159                    delta
160                })
161                .collect();
162            let mut observed = Vec::new();
163            let event = queue
164                .resolve_update(
165                    HummockVersionUpdate::VersionDeltas(deltas),
166                    |before, delta| {
167                        observed.push((before.id.as_raw_id(), delta.id.as_raw_id()));
168                    },
169                )
170                .unwrap();
171            assert_eq!(
172                observed,
173                vec![(id * 2 - 2, id * 2 - 1), (id * 2 - 1, id * 2)]
174            );
175            queue.enqueue(event, std::future::ready(()).boxed());
176        }
177        let first = queue.next_events().await;
178        assert_eq!(first.len(), 16);
179        assert_eq!(
180            first
181                .iter()
182                .map(|event| event.new_pinned_version.id().as_raw_id())
183                .collect::<Vec<_>>(),
184            (1..=16).map(|id| id * 2).collect::<Vec<_>>()
185        );
186        let last = queue.next_events().await;
187        assert_eq!(last.len(), 1);
188        assert_eq!(last[0].new_pinned_version.id().as_raw_id(), 34);
189    }
190}
risingwave_storage/hummock/event_handler/version_update.rs

risingwave_storage/hummock/event_handler/
version_update.rs
