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: u64,
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) -> u64 {
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 page_row_count = u64::try_from(page.proposal_count)
270        .map_err(|_| NnsProposalError::CollectionAccountingOverflow)?;
271    let pages_fetched = state
272        .pages_fetched
273        .checked_add(1)
274        .ok_or(NnsProposalError::CollectionAccountingOverflow)?;
275    let proposals_fetched = state
276        .proposals_fetched
277        .checked_add(page_row_count)
278        .ok_or(NnsProposalError::CollectionAccountingOverflow)?;
279    let next_before_proposal_id = (page_count == state.page_size)
280        .then(|| {
281            page.proposals
282                .iter()
283                .filter_map(|proposal| proposal.proposal_id)
284                .min()
285                .filter(|proposal_id| *proposal_id > 1)
286        })
287        .flatten();
288    let status = if next_before_proposal_id.is_none() {
289        NnsProposalCollectionStatus::Complete
290    } else if pages_fetched == state.max_pages {
291        NnsProposalCollectionStatus::PageLimitReached
292    } else {
293        NnsProposalCollectionStatus::Collecting
294    };
295    let next_state = NnsProposalCollectionState {
296        source: Some(page.context.source.clone()),
297        pages_fetched,
298        proposals_fetched,
299        next_before_proposal_id,
300        updated_at: request.fetched_at.clone(),
301        status,
302        ..state.clone()
303    };
304    validate_collection_state(&next_state)?;
305    Ok(NnsProposalCollectionStep {
306        page,
307        state: next_state,
308    })
309}
310
311fn validate_continuation_request(
312    request: &NnsGovernanceRequest,
313    state: &NnsProposalCollectionState,
314) -> Result<(), NnsProposalError> {
315    if request.network != state.network {
316        return Err(NnsProposalError::CollectionRequestMismatch {
317            field: "network",
318            expected: state.network.clone(),
319            actual: request.network.clone(),
320        });
321    }
322    if request.source != state.requested_source {
323        return Err(NnsProposalError::CollectionRequestMismatch {
324            field: "requested_source",
325            expected: format!("{:?}", state.requested_source),
326            actual: format!("{:?}", request.source),
327        });
328    }
329    Ok(())
330}
331
332pub(super) fn validate_collection_state(
333    state: &NnsProposalCollectionState,
334) -> Result<(), NnsProposalError> {
335    let invalid = |reason| NnsProposalError::InvalidCollectionState { reason };
336    if state.schema_version != NNS_PROPOSAL_COLLECTION_STATE_SCHEMA_VERSION {
337        return Err(invalid(format!(
338            "schema_version is {}, expected {NNS_PROPOSAL_COLLECTION_STATE_SCHEMA_VERSION}",
339            state.schema_version
340        )));
341    }
342    if state.governance_canister_id != MAINNET_GOVERNANCE_CANISTER_ID {
343        return Err(invalid(format!(
344            "governance_canister_id is {}, expected {MAINNET_GOVERNANCE_CANISTER_ID}",
345            state.governance_canister_id
346        )));
347    }
348    let state_request = NnsGovernanceRequest {
349        network: state.network.clone(),
350        fetched_at: state.started_at.clone(),
351        source: state.requested_source.clone(),
352    };
353    validate_governance_request(&state_request)?;
354    validate_proposal_page_size(state.page_size)?;
355    if state.max_pages == 0 {
356        return Err(invalid("max_pages must be greater than zero".to_string()));
357    }
358    if state.pages_fetched > state.max_pages {
359        return Err(invalid(format!(
360            "pages_fetched {} exceeds max_pages {}",
361            state.pages_fetched, state.max_pages
362        )));
363    }
364    if let Some(source) = &state.source {
365        validate_source_provenance(&state.requested_source, source)?;
366    }
367
368    let page_size = u64::from(state.page_size);
369    let pages_fetched = u64::from(state.pages_fetched);
370    let proposals_fetched = state.proposals_fetched;
371    let maximum_rows = pages_fetched
372        .checked_mul(page_size)
373        .ok_or(NnsProposalError::CollectionAccountingOverflow)?;
374    let minimum_rows = pages_fetched
375        .saturating_sub(1)
376        .checked_mul(page_size)
377        .ok_or(NnsProposalError::CollectionAccountingOverflow)?;
378    if proposals_fetched < minimum_rows || proposals_fetched > maximum_rows {
379        return Err(invalid(format!(
380            "proposals_fetched {} is outside {}..={} for {} pages of size {}",
381            state.proposals_fetched,
382            minimum_rows,
383            maximum_rows,
384            state.pages_fetched,
385            state.page_size
386        )));
387    }
388    if state
389        .next_before_proposal_id
390        .is_some_and(|cursor| cursor <= 1)
391    {
392        return Err(invalid(
393            "next_before_proposal_id must be greater than one".to_string(),
394        ));
395    }
396
397    let valid_lifecycle = match state.status {
398        NnsProposalCollectionStatus::Ready => {
399            state.pages_fetched == 0
400                && state.proposals_fetched == 0
401                && state.next_before_proposal_id.is_none()
402                && state.source.is_none()
403        }
404        NnsProposalCollectionStatus::Collecting => {
405            state.pages_fetched > 0
406                && state.pages_fetched < state.max_pages
407                && proposals_fetched == maximum_rows
408                && state.next_before_proposal_id.is_some()
409                && state.source.is_some()
410        }
411        NnsProposalCollectionStatus::Complete => {
412            state.pages_fetched > 0
413                && state.next_before_proposal_id.is_none()
414                && state.source.is_some()
415        }
416        NnsProposalCollectionStatus::PageLimitReached => {
417            state.pages_fetched == state.max_pages
418                && proposals_fetched == maximum_rows
419                && state.next_before_proposal_id.is_some()
420                && state.source.is_some()
421        }
422    };
423    if !valid_lifecycle {
424        return Err(invalid(format!(
425            "status {} disagrees with cursor, provenance, or counters",
426            state.status
427        )));
428    }
429    Ok(())
430}
ic_query/nns/proposals/report/collection.rs

ic_query/nns/proposals/report/
collection.rs
