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

ic_query/nns/proposals/report/
collection.rs
