ic_query/nns/proposals/report/
collection.rs

1//! Module: nns::proposals::report::collection
2//!
3//! Responsibility: advance caller-persisted complete proposal walks one page at a time.
4//! Does not own: stable memory, filesystem caches, scheduling, retries, or publication.
5//! Boundary: validates resumable state before and after exactly one bounded source call.
6
7#[cfg(feature = "nns-host")]
8use super::NnsProposalHostError;
9use super::{
10    NnsProposalError,
11    model::{NnsProposalListReport, NnsProposalListRequest},
12    source::{
13        NnsProposalSource, build_nns_proposal_list_report_with_source, validate_proposal_page_size,
14    },
15};
16use crate::nns::{
17    MAINNET_GOVERNANCE_CANISTER_ID,
18    governance::{
19        NnsGovernanceRequest, NnsGovernanceSourceProvenance, NnsGovernanceSourceSelection,
20        validate_governance_request, validate_source_provenance,
21    },
22};
23#[cfg(feature = "nns-host")]
24use crate::{nns::LiveNnsSource, runtime::block_on_current_thread};
25use serde::{Deserialize, Serialize};
26use std::fmt;
27
28/// Version of the persistable resumable NNS proposal collection state.
29pub const NNS_PROPOSAL_COLLECTION_STATE_SCHEMA_VERSION: u32 = 1;
30
31///
32/// NnsProposalCollectionStatus
33///
34/// Lifecycle of a caller-owned resumable proposal collection.
35///
36
37#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
38#[serde(rename_all = "snake_case")]
39pub enum NnsProposalCollectionStatus {
40    /// No source page has been admitted yet.
41    Ready,
42    /// Another bounded page may be requested.
43    Collecting,
44    /// Governance API exhaustion was observed.
45    Complete,
46    /// Another cursor exists, but the configured page ceiling was consumed.
47    PageLimitReached,
48}
49
50impl NnsProposalCollectionStatus {
51    /// Return the stable JSON and display label.
52    #[must_use]
53    pub const fn as_str(self) -> &'static str {
54        match self {
55            Self::Ready => "ready",
56            Self::Collecting => "collecting",
57            Self::Complete => "complete",
58            Self::PageLimitReached => "page_limit_reached",
59        }
60    }
61}
62
63impl fmt::Display for NnsProposalCollectionStatus {
64    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
65        formatter.write_str(self.as_str())
66    }
67}
68
69///
70/// NnsProposalCollectionState
71///
72/// Serializable continuation state for an explicitly bounded proposal walk.
73///
74
75#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
76pub struct NnsProposalCollectionState {
77    schema_version: u32,
78    network: String,
79    governance_canister_id: String,
80    requested_source: NnsGovernanceSourceSelection,
81    source: Option<NnsGovernanceSourceProvenance>,
82    page_size: u32,
83    max_pages: u32,
84    pages_fetched: u32,
85    proposals_fetched: usize,
86    next_before_proposal_id: Option<u64>,
87    started_at: String,
88    updated_at: String,
89    status: NnsProposalCollectionStatus,
90}
91
92impl NnsProposalCollectionState {
93    /// Start an empty proposal walk with explicit per-page and cumulative call ceilings.
94    pub fn new(
95        request: &NnsGovernanceRequest,
96        page_size: u32,
97        max_pages: u32,
98    ) -> Result<Self, NnsProposalError> {
99        validate_governance_request(request)?;
100        validate_proposal_page_size(page_size)?;
101        if max_pages == 0 {
102            return Err(NnsProposalError::InvalidCollectionMaxPages);
103        }
104        Ok(Self {
105            schema_version: NNS_PROPOSAL_COLLECTION_STATE_SCHEMA_VERSION,
106            network: request.network.clone(),
107            governance_canister_id: MAINNET_GOVERNANCE_CANISTER_ID.to_string(),
108            requested_source: request.source.clone(),
109            source: None,
110            page_size,
111            max_pages,
112            pages_fetched: 0,
113            proposals_fetched: 0,
114            next_before_proposal_id: None,
115            started_at: request.fetched_at.clone(),
116            updated_at: request.fetched_at.clone(),
117            status: NnsProposalCollectionStatus::Ready,
118        })
119    }
120
121    /// Return the state schema version.
122    #[must_use]
123    pub const fn schema_version(&self) -> u32 {
124        self.schema_version
125    }
126
127    /// Return the fixed network identity.
128    #[must_use]
129    pub fn network(&self) -> &str {
130        &self.network
131    }
132
133    /// Return the fixed Governance canister identity.
134    #[must_use]
135    pub fn governance_canister_id(&self) -> &str {
136        &self.governance_canister_id
137    }
138
139    /// Return the source selection fixed when collection started.
140    #[must_use]
141    pub const fn requested_source(&self) -> &NnsGovernanceSourceSelection {
142        &self.requested_source
143    }
144
145    /// Return the concrete source provenance after the first admitted page.
146    #[must_use]
147    pub const fn source(&self) -> Option<&NnsGovernanceSourceProvenance> {
148        self.source.as_ref()
149    }
150
151    /// Return the maximum rows requested from each Governance call.
152    #[must_use]
153    pub const fn page_size(&self) -> u32 {
154        self.page_size
155    }
156
157    /// Return the cumulative source-call ceiling.
158    #[must_use]
159    pub const fn max_pages(&self) -> u32 {
160        self.max_pages
161    }
162
163    /// Return the number of successfully admitted pages.
164    #[must_use]
165    pub const fn pages_fetched(&self) -> u32 {
166        self.pages_fetched
167    }
168
169    /// Return the number of successfully admitted proposal rows.
170    #[must_use]
171    pub const fn proposals_fetched(&self) -> usize {
172        self.proposals_fetched
173    }
174
175    /// Return the exclusive cursor for the next page, when one remains.
176    #[must_use]
177    pub const fn next_before_proposal_id(&self) -> Option<u64> {
178        self.next_before_proposal_id
179    }
180
181    /// Return the caller-supplied time at which the collection state was created.
182    #[must_use]
183    pub fn started_at(&self) -> &str {
184        &self.started_at
185    }
186
187    /// Return the caller-supplied time attached to the latest admitted page.
188    #[must_use]
189    pub fn updated_at(&self) -> &str {
190        &self.updated_at
191    }
192
193    /// Return the collection lifecycle status.
194    #[must_use]
195    pub const fn status(&self) -> NnsProposalCollectionStatus {
196        self.status
197    }
198
199    /// Return whether Governance API exhaustion was observed.
200    #[must_use]
201    pub const fn is_complete(&self) -> bool {
202        matches!(self.status, NnsProposalCollectionStatus::Complete)
203    }
204}
205
206///
207/// NnsProposalCollectionStep
208///
209/// One admitted bounded page and the continuation state that follows it.
210///
211
212#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
213pub struct NnsProposalCollectionStep {
214    /// Unfiltered page returned by the shared proposal report builder.
215    pub page: NnsProposalListReport,
216    /// Validated state to persist only after retaining the page.
217    pub state: NnsProposalCollectionState,
218}
219
220/// Advance a resumable proposal walk through the native replica adapter.
221#[cfg(feature = "nns-host")]
222pub fn advance_nns_proposal_collection(
223    request: &NnsGovernanceRequest,
224    state: &NnsProposalCollectionState,
225) -> Result<NnsProposalCollectionStep, NnsProposalHostError> {
226    Ok(block_on_current_thread(
227        advance_nns_proposal_collection_with_source(request, state, &LiveNnsSource),
228    )??)
229}
230
231/// Advance a resumable proposal walk by exactly one caller-runtime source call.
232pub async fn advance_nns_proposal_collection_with_source(
233    request: &NnsGovernanceRequest,
234    state: &NnsProposalCollectionState,
235    source: &dyn NnsProposalSource,
236) -> Result<NnsProposalCollectionStep, NnsProposalError> {
237    validate_governance_request(request)?;
238    validate_collection_state(state)?;
239    validate_continuation_request(request, state)?;
240    match state.status {
241        NnsProposalCollectionStatus::Complete => {
242            return Err(NnsProposalError::CollectionComplete {
243                pages_fetched: state.pages_fetched,
244            });
245        }
246        NnsProposalCollectionStatus::PageLimitReached => {
247            return Err(NnsProposalError::CollectionPageLimitReached {
248                pages_fetched: state.pages_fetched,
249                max_pages: state.max_pages,
250            });
251        }
252        NnsProposalCollectionStatus::Ready | NnsProposalCollectionStatus::Collecting => {}
253    }
254
255    let mut page_request = NnsProposalListRequest::new(request.clone(), state.page_size);
256    page_request.before_proposal_id = state.next_before_proposal_id;
257    let page = build_nns_proposal_list_report_with_source(&page_request, source).await?;
258    if let Some(expected) = &state.source
259        && *expected != page.context.source
260    {
261        return Err(NnsProposalError::CollectionSourceChanged {
262            expected: expected.clone(),
263            actual: page.context.source,
264        });
265    }
266
267    let page_count = u32::try_from(page.proposal_count)
268        .map_err(|_| NnsProposalError::CollectionAccountingOverflow)?;
269    let pages_fetched = state
270        .pages_fetched
271        .checked_add(1)
272        .ok_or(NnsProposalError::CollectionAccountingOverflow)?;
273    let proposals_fetched = state
274        .proposals_fetched
275        .checked_add(page.proposal_count)
276        .ok_or(NnsProposalError::CollectionAccountingOverflow)?;
277    let next_before_proposal_id = (page_count == state.page_size)
278        .then(|| {
279            page.proposals
280                .iter()
281                .filter_map(|proposal| proposal.proposal_id)
282                .min()
283                .filter(|proposal_id| *proposal_id > 1)
284        })
285        .flatten();
286    let status = if next_before_proposal_id.is_none() {
287        NnsProposalCollectionStatus::Complete
288    } else if pages_fetched == state.max_pages {
289        NnsProposalCollectionStatus::PageLimitReached
290    } else {
291        NnsProposalCollectionStatus::Collecting
292    };
293    let next_state = NnsProposalCollectionState {
294        source: Some(page.context.source.clone()),
295        pages_fetched,
296        proposals_fetched,
297        next_before_proposal_id,
298        updated_at: request.fetched_at.clone(),
299        status,
300        ..state.clone()
301    };
302    validate_collection_state(&next_state)?;
303    Ok(NnsProposalCollectionStep {
304        page,
305        state: next_state,
306    })
307}
308
309fn validate_continuation_request(
310    request: &NnsGovernanceRequest,
311    state: &NnsProposalCollectionState,
312) -> Result<(), NnsProposalError> {
313    if request.network != state.network {
314        return Err(NnsProposalError::CollectionRequestMismatch {
315            field: "network",
316            expected: state.network.clone(),
317            actual: request.network.clone(),
318        });
319    }
320    if request.source != state.requested_source {
321        return Err(NnsProposalError::CollectionRequestMismatch {
322            field: "requested_source",
323            expected: format!("{:?}", state.requested_source),
324            actual: format!("{:?}", request.source),
325        });
326    }
327    Ok(())
328}
329
330pub(super) fn validate_collection_state(
331    state: &NnsProposalCollectionState,
332) -> Result<(), NnsProposalError> {
333    let invalid = |reason| NnsProposalError::InvalidCollectionState { reason };
334    if state.schema_version != NNS_PROPOSAL_COLLECTION_STATE_SCHEMA_VERSION {
335        return Err(invalid(format!(
336            "schema_version is {}, expected {NNS_PROPOSAL_COLLECTION_STATE_SCHEMA_VERSION}",
337            state.schema_version
338        )));
339    }
340    if state.governance_canister_id != MAINNET_GOVERNANCE_CANISTER_ID {
341        return Err(invalid(format!(
342            "governance_canister_id is {}, expected {MAINNET_GOVERNANCE_CANISTER_ID}",
343            state.governance_canister_id
344        )));
345    }
346    let state_request = NnsGovernanceRequest {
347        network: state.network.clone(),
348        fetched_at: state.started_at.clone(),
349        source: state.requested_source.clone(),
350    };
351    validate_governance_request(&state_request)?;
352    validate_proposal_page_size(state.page_size)?;
353    if state.max_pages == 0 {
354        return Err(invalid("max_pages must be greater than zero".to_string()));
355    }
356    if state.pages_fetched > state.max_pages {
357        return Err(invalid(format!(
358            "pages_fetched {} exceeds max_pages {}",
359            state.pages_fetched, state.max_pages
360        )));
361    }
362    if let Some(source) = &state.source {
363        validate_source_provenance(&state.requested_source, source)?;
364    }
365
366    let page_size = u64::from(state.page_size);
367    let pages_fetched = u64::from(state.pages_fetched);
368    let proposals_fetched = u64::try_from(state.proposals_fetched)
369        .map_err(|_| NnsProposalError::CollectionAccountingOverflow)?;
370    let maximum_rows = pages_fetched
371        .checked_mul(page_size)
372        .ok_or(NnsProposalError::CollectionAccountingOverflow)?;
373    let minimum_rows = pages_fetched
374        .saturating_sub(1)
375        .checked_mul(page_size)
376        .ok_or(NnsProposalError::CollectionAccountingOverflow)?;
377    if proposals_fetched < minimum_rows || proposals_fetched > maximum_rows {
378        return Err(invalid(format!(
379            "proposals_fetched {} is outside {}..={} for {} pages of size {}",
380            state.proposals_fetched,
381            minimum_rows,
382            maximum_rows,
383            state.pages_fetched,
384            state.page_size
385        )));
386    }
387    if state
388        .next_before_proposal_id
389        .is_some_and(|cursor| cursor <= 1)
390    {
391        return Err(invalid(
392            "next_before_proposal_id must be greater than one".to_string(),
393        ));
394    }
395
396    let valid_lifecycle = match state.status {
397        NnsProposalCollectionStatus::Ready => {
398            state.pages_fetched == 0
399                && state.proposals_fetched == 0
400                && state.next_before_proposal_id.is_none()
401                && state.source.is_none()
402        }
403        NnsProposalCollectionStatus::Collecting => {
404            state.pages_fetched > 0
405                && state.pages_fetched < state.max_pages
406                && proposals_fetched == maximum_rows
407                && state.next_before_proposal_id.is_some()
408                && state.source.is_some()
409        }
410        NnsProposalCollectionStatus::Complete => {
411            state.pages_fetched > 0
412                && state.next_before_proposal_id.is_none()
413                && state.source.is_some()
414        }
415        NnsProposalCollectionStatus::PageLimitReached => {
416            state.pages_fetched == state.max_pages
417                && proposals_fetched == maximum_rows
418                && state.next_before_proposal_id.is_some()
419                && state.source.is_some()
420        }
421    };
422    if !valid_lifecycle {
423        return Err(invalid(format!(
424            "status {} disagrees with cursor, provenance, or counters",
425            state.status
426        )));
427    }
428    Ok(())
429}
ic_query/nns/proposals/report/collection.rs

ic_query/nns/proposals/report/
collection.rs
