1#![warn(missing_docs)]
48#![warn(unused_extern_crates)]
49
50mod metrics;
51
52pub use sp_statement_store::{Error, StatementStore, MAX_TOPICS};
53
54use metrics::MetricsLink as PrometheusMetrics;
55use parking_lot::RwLock;
56use prometheus_endpoint::Registry as PrometheusRegistry;
57use sc_keystore::LocalKeystore;
58use sp_api::ProvideRuntimeApi;
59use sp_blockchain::HeaderBackend;
60use sp_core::{crypto::UncheckedFrom, hexdisplay::HexDisplay, traits::SpawnNamed, Decode, Encode};
61use sp_runtime::traits::Block as BlockT;
62use sp_statement_store::{
63 runtime_api::{
64 InvalidStatement, StatementSource, StatementStoreExt, ValidStatement, ValidateStatement,
65 },
66 AccountId, BlockHash, Channel, DecryptionKey, Hash, NetworkPriority, Proof, Result, Statement,
67 SubmitResult, Topic,
68};
69use std::{
70 collections::{BTreeMap, HashMap, HashSet},
71 sync::Arc,
72};
73
74const KEY_VERSION: &[u8] = b"version".as_slice();
75const CURRENT_VERSION: u32 = 1;
76
77const LOG_TARGET: &str = "statement-store";
78
79const DEFAULT_PURGE_AFTER_SEC: u64 = 2 * 24 * 60 * 60; const DEFAULT_MAX_TOTAL_STATEMENTS: usize = 8192;
81const DEFAULT_MAX_TOTAL_SIZE: usize = 64 * 1024 * 1024;
82
83const MAINTENANCE_PERIOD: std::time::Duration = std::time::Duration::from_secs(30);
84
85mod col {
86 pub const META: u8 = 0;
87 pub const STATEMENTS: u8 = 1;
88 pub const EXPIRED: u8 = 2;
89
90 pub const COUNT: u8 = 3;
91}
92
93#[derive(Eq, PartialEq, Debug, Ord, PartialOrd, Clone, Copy)]
94struct Priority(u32);
95
96#[derive(PartialEq, Eq)]
97struct PriorityKey {
98 hash: Hash,
99 priority: Priority,
100}
101
102impl PartialOrd for PriorityKey {
103 fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
104 Some(self.cmp(other))
105 }
106}
107
108impl Ord for PriorityKey {
109 fn cmp(&self, other: &Self) -> std::cmp::Ordering {
110 self.priority.cmp(&other.priority).then_with(|| self.hash.cmp(&other.hash))
111 }
112}
113
114#[derive(PartialEq, Eq)]
115struct ChannelEntry {
116 hash: Hash,
117 priority: Priority,
118}
119
120#[derive(Default)]
121struct StatementsForAccount {
122 by_priority: BTreeMap<PriorityKey, (Option<Channel>, usize)>,
124 channels: HashMap<Channel, ChannelEntry>,
126 data_size: usize,
128}
129
130pub struct Options {
132 max_total_statements: usize,
135 max_total_size: usize,
138 purge_after_sec: u64,
140}
141
142impl Default for Options {
143 fn default() -> Self {
144 Options {
145 max_total_statements: DEFAULT_MAX_TOTAL_STATEMENTS,
146 max_total_size: DEFAULT_MAX_TOTAL_SIZE,
147 purge_after_sec: DEFAULT_PURGE_AFTER_SEC,
148 }
149 }
150}
151
152#[derive(Default)]
153struct Index {
154 by_topic: HashMap<Topic, HashSet<Hash>>,
155 by_dec_key: HashMap<Option<DecryptionKey>, HashSet<Hash>>,
156 topics_and_keys: HashMap<Hash, ([Option<Topic>; MAX_TOPICS], Option<DecryptionKey>)>,
157 entries: HashMap<Hash, (AccountId, Priority, usize)>,
158 expired: HashMap<Hash, u64>, accounts: HashMap<AccountId, StatementsForAccount>,
160 options: Options,
161 total_size: usize,
162}
163
164struct ClientWrapper<Block, Client> {
165 client: Arc<Client>,
166 _block: std::marker::PhantomData<Block>,
167}
168
169impl<Block, Client> ClientWrapper<Block, Client>
170where
171 Block: BlockT,
172 Block::Hash: From<BlockHash>,
173 Client: ProvideRuntimeApi<Block> + HeaderBackend<Block> + Send + Sync + 'static,
174 Client::Api: ValidateStatement<Block>,
175{
176 fn validate_statement(
177 &self,
178 block: Option<BlockHash>,
179 source: StatementSource,
180 statement: Statement,
181 ) -> std::result::Result<ValidStatement, InvalidStatement> {
182 let api = self.client.runtime_api();
183 let block = block.map(Into::into).unwrap_or_else(|| {
184 self.client.info().finalized_hash
186 });
187 api.validate_statement(block, source, statement)
188 .map_err(|_| InvalidStatement::InternalError)?
189 }
190}
191
192pub struct Store {
194 db: parity_db::Db,
195 index: RwLock<Index>,
196 validate_fn: Box<
197 dyn Fn(
198 Option<BlockHash>,
199 StatementSource,
200 Statement,
201 ) -> std::result::Result<ValidStatement, InvalidStatement>
202 + Send
203 + Sync,
204 >,
205 keystore: Arc<LocalKeystore>,
206 time_override: Option<u64>,
208 metrics: PrometheusMetrics,
209}
210
211enum IndexQuery {
212 Unknown,
213 Exists,
214 Expired,
215}
216
217enum MaybeInserted {
218 Inserted(HashSet<Hash>),
219 Ignored,
220}
221
222impl Index {
223 fn new(options: Options) -> Index {
224 Index { options, ..Default::default() }
225 }
226
227 fn insert_new(&mut self, hash: Hash, account: AccountId, statement: &Statement) {
228 let mut all_topics = [None; MAX_TOPICS];
229 let mut nt = 0;
230 while let Some(t) = statement.topic(nt) {
231 self.by_topic.entry(t).or_default().insert(hash);
232 all_topics[nt] = Some(t);
233 nt += 1;
234 }
235 let key = statement.decryption_key();
236 self.by_dec_key.entry(key).or_default().insert(hash);
237 if nt > 0 || key.is_some() {
238 self.topics_and_keys.insert(hash, (all_topics, key));
239 }
240 let priority = Priority(statement.priority().unwrap_or(0));
241 self.entries.insert(hash, (account, priority, statement.data_len()));
242 self.total_size += statement.data_len();
243 let account_info = self.accounts.entry(account).or_default();
244 account_info.data_size += statement.data_len();
245 if let Some(channel) = statement.channel() {
246 account_info.channels.insert(channel, ChannelEntry { hash, priority });
247 }
248 account_info
249 .by_priority
250 .insert(PriorityKey { hash, priority }, (statement.channel(), statement.data_len()));
251 }
252
253 fn query(&self, hash: &Hash) -> IndexQuery {
254 if self.entries.contains_key(hash) {
255 return IndexQuery::Exists
256 }
257 if self.expired.contains_key(hash) {
258 return IndexQuery::Expired
259 }
260 IndexQuery::Unknown
261 }
262
263 fn insert_expired(&mut self, hash: Hash, timestamp: u64) {
264 self.expired.insert(hash, timestamp);
265 }
266
267 fn iterate_with(
268 &self,
269 key: Option<DecryptionKey>,
270 match_all_topics: &[Topic],
271 mut f: impl FnMut(&Hash) -> Result<()>,
272 ) -> Result<()> {
273 let empty = HashSet::new();
274 let mut sets: [&HashSet<Hash>; MAX_TOPICS + 1] = [∅ MAX_TOPICS + 1];
275 if match_all_topics.len() > MAX_TOPICS {
276 return Ok(())
277 }
278 let key_set = self.by_dec_key.get(&key);
279 if key_set.map_or(0, |s| s.len()) == 0 {
280 return Ok(())
282 }
283 sets[0] = key_set.expect("Function returns if key_set is None");
284 for (i, t) in match_all_topics.iter().enumerate() {
285 let set = self.by_topic.get(t);
286 if set.map_or(0, |s| s.len()) == 0 {
287 return Ok(())
289 }
290 sets[i + 1] = set.expect("Function returns if set is None");
291 }
292 let sets = &mut sets[0..match_all_topics.len() + 1];
293 sets.sort_by_key(|s| s.len());
295 for item in sets[0] {
296 if sets[1..].iter().all(|set| set.contains(item)) {
297 log::trace!(
298 target: LOG_TARGET,
299 "Iterating by topic/key: statement {:?}",
300 HexDisplay::from(item)
301 );
302 f(item)?
303 }
304 }
305 Ok(())
306 }
307
308 fn maintain(&mut self, current_time: u64) -> Vec<Hash> {
309 let mut purged = Vec::new();
311 self.expired.retain(|hash, timestamp| {
312 if *timestamp + self.options.purge_after_sec <= current_time {
313 purged.push(*hash);
314 log::trace!(target: LOG_TARGET, "Purged statement {:?}", HexDisplay::from(hash));
315 false
316 } else {
317 true
318 }
319 });
320 purged
321 }
322
323 fn make_expired(&mut self, hash: &Hash, current_time: u64) -> bool {
324 if let Some((account, priority, len)) = self.entries.remove(hash) {
325 self.total_size -= len;
326 if let Some((topics, key)) = self.topics_and_keys.remove(hash) {
327 for t in topics.into_iter().flatten() {
328 if let std::collections::hash_map::Entry::Occupied(mut set) =
329 self.by_topic.entry(t)
330 {
331 set.get_mut().remove(hash);
332 if set.get().is_empty() {
333 set.remove_entry();
334 }
335 }
336 }
337 if let std::collections::hash_map::Entry::Occupied(mut set) =
338 self.by_dec_key.entry(key)
339 {
340 set.get_mut().remove(hash);
341 if set.get().is_empty() {
342 set.remove_entry();
343 }
344 }
345 }
346 self.expired.insert(*hash, current_time);
347 if let std::collections::hash_map::Entry::Occupied(mut account_rec) =
348 self.accounts.entry(account)
349 {
350 let key = PriorityKey { hash: *hash, priority };
351 if let Some((channel, len)) = account_rec.get_mut().by_priority.remove(&key) {
352 account_rec.get_mut().data_size -= len;
353 if let Some(channel) = channel {
354 account_rec.get_mut().channels.remove(&channel);
355 }
356 }
357 if account_rec.get().by_priority.is_empty() {
358 account_rec.remove_entry();
359 }
360 }
361 log::trace!(target: LOG_TARGET, "Expired statement {:?}", HexDisplay::from(hash));
362 true
363 } else {
364 false
365 }
366 }
367
368 fn insert(
369 &mut self,
370 hash: Hash,
371 statement: &Statement,
372 account: &AccountId,
373 validation: &ValidStatement,
374 current_time: u64,
375 ) -> MaybeInserted {
376 let statement_len = statement.data_len();
377 if statement_len > validation.max_size as usize {
378 log::debug!(
379 target: LOG_TARGET,
380 "Ignored oversize message: {:?} ({} bytes)",
381 HexDisplay::from(&hash),
382 statement_len,
383 );
384 return MaybeInserted::Ignored
385 }
386
387 let mut evicted = HashSet::new();
388 let mut would_free_size = 0;
389 let priority = Priority(statement.priority().unwrap_or(0));
390 let (max_size, max_count) = (validation.max_size as usize, validation.max_count as usize);
391 if let Some(account_rec) = self.accounts.get(account) {
395 if let Some(channel) = statement.channel() {
396 if let Some(channel_record) = account_rec.channels.get(&channel) {
397 if priority <= channel_record.priority {
398 log::debug!(
400 target: LOG_TARGET,
401 "Ignored lower priority channel message: {:?} {:?} <= {:?}",
402 HexDisplay::from(&hash),
403 priority,
404 channel_record.priority,
405 );
406 return MaybeInserted::Ignored
407 } else {
408 log::debug!(
411 target: LOG_TARGET,
412 "Replacing higher priority channel message: {:?} ({:?}) > {:?} ({:?})",
413 HexDisplay::from(&hash),
414 priority,
415 HexDisplay::from(&channel_record.hash),
416 channel_record.priority,
417 );
418 let key = PriorityKey {
419 hash: channel_record.hash,
420 priority: channel_record.priority,
421 };
422 if let Some((_channel, len)) = account_rec.by_priority.get(&key) {
423 would_free_size += *len;
424 evicted.insert(channel_record.hash);
425 }
426 }
427 }
428 }
429 for (entry, (_, len)) in account_rec.by_priority.iter() {
431 if (account_rec.data_size - would_free_size + statement_len <= max_size) &&
432 account_rec.by_priority.len() + 1 - evicted.len() <= max_count
433 {
434 break
436 }
437 if evicted.contains(&entry.hash) {
438 continue
440 }
441 if entry.priority >= priority {
442 log::debug!(
443 target: LOG_TARGET,
444 "Ignored message due to constraints {:?} {:?} < {:?}",
445 HexDisplay::from(&hash),
446 priority,
447 entry.priority,
448 );
449 return MaybeInserted::Ignored
450 }
451 evicted.insert(entry.hash);
452 would_free_size += len;
453 }
454 }
455 if !((self.total_size - would_free_size + statement_len <= self.options.max_total_size) &&
457 self.entries.len() + 1 - evicted.len() <= self.options.max_total_statements)
458 {
459 log::debug!(
460 target: LOG_TARGET,
461 "Ignored statement {} because the store is full (size={}, count={})",
462 HexDisplay::from(&hash),
463 self.total_size,
464 self.entries.len(),
465 );
466 return MaybeInserted::Ignored
467 }
468
469 for h in &evicted {
470 self.make_expired(h, current_time);
471 }
472 self.insert_new(hash, *account, statement);
473 MaybeInserted::Inserted(evicted)
474 }
475}
476
477impl Store {
478 pub fn new_shared<Block, Client>(
481 path: &std::path::Path,
482 options: Options,
483 client: Arc<Client>,
484 keystore: Arc<LocalKeystore>,
485 prometheus: Option<&PrometheusRegistry>,
486 task_spawner: &dyn SpawnNamed,
487 ) -> Result<Arc<Store>>
488 where
489 Block: BlockT,
490 Block::Hash: From<BlockHash>,
491 Client: ProvideRuntimeApi<Block>
492 + HeaderBackend<Block>
493 + sc_client_api::ExecutorProvider<Block>
494 + Send
495 + Sync
496 + 'static,
497 Client::Api: ValidateStatement<Block>,
498 {
499 let store = Arc::new(Self::new(path, options, client, keystore, prometheus)?);
500
501 let worker_store = store.clone();
503 task_spawner.spawn(
504 "statement-store-maintenance",
505 Some("statement-store"),
506 Box::pin(async move {
507 let mut interval = tokio::time::interval(MAINTENANCE_PERIOD);
508 loop {
509 interval.tick().await;
510 worker_store.maintain();
511 }
512 }),
513 );
514
515 Ok(store)
516 }
517
518 fn new<Block, Client>(
521 path: &std::path::Path,
522 options: Options,
523 client: Arc<Client>,
524 keystore: Arc<LocalKeystore>,
525 prometheus: Option<&PrometheusRegistry>,
526 ) -> Result<Store>
527 where
528 Block: BlockT,
529 Block::Hash: From<BlockHash>,
530 Client: ProvideRuntimeApi<Block> + HeaderBackend<Block> + Send + Sync + 'static,
531 Client::Api: ValidateStatement<Block>,
532 {
533 let mut path: std::path::PathBuf = path.into();
534 path.push("statements");
535
536 let mut config = parity_db::Options::with_columns(&path, col::COUNT);
537
538 let statement_col = &mut config.columns[col::STATEMENTS as usize];
539 statement_col.ref_counted = false;
540 statement_col.preimage = true;
541 statement_col.uniform = true;
542 let db = parity_db::Db::open_or_create(&config).map_err(|e| Error::Db(e.to_string()))?;
543 match db.get(col::META, &KEY_VERSION).map_err(|e| Error::Db(e.to_string()))? {
544 Some(version) => {
545 let version = u32::from_le_bytes(
546 version
547 .try_into()
548 .map_err(|_| Error::Db("Error reading database version".into()))?,
549 );
550 if version != CURRENT_VERSION {
551 return Err(Error::Db(format!("Unsupported database version: {version}")))
552 }
553 },
554 None => {
555 db.commit([(
556 col::META,
557 KEY_VERSION.to_vec(),
558 Some(CURRENT_VERSION.to_le_bytes().to_vec()),
559 )])
560 .map_err(|e| Error::Db(e.to_string()))?;
561 },
562 }
563
564 let validator = ClientWrapper { client, _block: Default::default() };
565 let validate_fn = Box::new(move |block, source, statement| {
566 validator.validate_statement(block, source, statement)
567 });
568
569 let store = Store {
570 db,
571 index: RwLock::new(Index::new(options)),
572 validate_fn,
573 keystore,
574 time_override: None,
575 metrics: PrometheusMetrics::new(prometheus),
576 };
577 store.populate()?;
578 Ok(store)
579 }
580
581 fn populate(&self) -> Result<()> {
586 {
587 let mut index = self.index.write();
588 self.db
589 .iter_column_while(col::STATEMENTS, |item| {
590 let statement = item.value;
591 if let Ok(statement) = Statement::decode(&mut statement.as_slice()) {
592 let hash = statement.hash();
593 log::trace!(
594 target: LOG_TARGET,
595 "Statement loaded {:?}",
596 HexDisplay::from(&hash)
597 );
598 if let Some(account_id) = statement.account_id() {
599 index.insert_new(hash, account_id, &statement);
600 } else {
601 log::debug!(
602 target: LOG_TARGET,
603 "Error decoding statement loaded from the DB: {:?}",
604 HexDisplay::from(&hash)
605 );
606 }
607 }
608 true
609 })
610 .map_err(|e| Error::Db(e.to_string()))?;
611 self.db
612 .iter_column_while(col::EXPIRED, |item| {
613 let expired_info = item.value;
614 if let Ok((hash, timestamp)) =
615 <(Hash, u64)>::decode(&mut expired_info.as_slice())
616 {
617 log::trace!(
618 target: LOG_TARGET,
619 "Statement loaded (expired): {:?}",
620 HexDisplay::from(&hash)
621 );
622 index.insert_expired(hash, timestamp);
623 }
624 true
625 })
626 .map_err(|e| Error::Db(e.to_string()))?;
627 }
628
629 self.maintain();
630 Ok(())
631 }
632
633 fn collect_statements<R>(
634 &self,
635 key: Option<DecryptionKey>,
636 match_all_topics: &[Topic],
637 mut f: impl FnMut(Statement) -> Option<R>,
638 ) -> Result<Vec<R>> {
639 let mut result = Vec::new();
640 let index = self.index.read();
641 index.iterate_with(key, match_all_topics, |hash| {
642 match self.db.get(col::STATEMENTS, hash).map_err(|e| Error::Db(e.to_string()))? {
643 Some(entry) => {
644 if let Ok(statement) = Statement::decode(&mut entry.as_slice()) {
645 if let Some(data) = f(statement) {
646 result.push(data);
647 }
648 } else {
649 log::warn!(
651 target: LOG_TARGET,
652 "Corrupt statement {:?}",
653 HexDisplay::from(hash)
654 );
655 }
656 },
657 None => {
658 log::warn!(
660 target: LOG_TARGET,
661 "Missing statement {:?}",
662 HexDisplay::from(hash)
663 );
664 },
665 }
666 Ok(())
667 })?;
668 Ok(result)
669 }
670
671 pub fn maintain(&self) {
673 log::trace!(target: LOG_TARGET, "Started store maintenance");
674 let deleted = self.index.write().maintain(self.timestamp());
675 let deleted: Vec<_> =
676 deleted.into_iter().map(|hash| (col::EXPIRED, hash.to_vec(), None)).collect();
677 let count = deleted.len() as u64;
678 if let Err(e) = self.db.commit(deleted) {
679 log::warn!(target: LOG_TARGET, "Error writing to the statement database: {:?}", e);
680 } else {
681 self.metrics.report(|metrics| metrics.statements_pruned.inc_by(count));
682 }
683 log::trace!(
684 target: LOG_TARGET,
685 "Completed store maintenance. Purged: {}, Active: {}, Expired: {}",
686 count,
687 self.index.read().entries.len(),
688 self.index.read().expired.len()
689 );
690 }
691
692 fn timestamp(&self) -> u64 {
693 self.time_override.unwrap_or_else(|| {
694 std::time::SystemTime::now()
695 .duration_since(std::time::UNIX_EPOCH)
696 .unwrap_or_default()
697 .as_secs()
698 })
699 }
700
701 #[cfg(test)]
702 fn set_time(&mut self, time: u64) {
703 self.time_override = Some(time);
704 }
705
706 pub fn as_statement_store_ext(self: Arc<Self>) -> StatementStoreExt {
708 StatementStoreExt::new(self)
709 }
710
711 fn posted_clear_inner<R>(
714 &self,
715 match_all_topics: &[Topic],
716 dest: [u8; 32],
717 mut map_f: impl FnMut(Statement, Vec<u8>) -> R,
719 ) -> Result<Vec<R>> {
720 self.collect_statements(Some(dest), match_all_topics, |statement| {
721 if let (Some(key), Some(_)) = (statement.decryption_key(), statement.data()) {
722 let public: sp_core::ed25519::Public = UncheckedFrom::unchecked_from(key);
723 let public: sp_statement_store::ed25519::Public = public.into();
724 match self.keystore.key_pair::<sp_statement_store::ed25519::Pair>(&public) {
725 Err(e) => {
726 log::debug!(
727 target: LOG_TARGET,
728 "Keystore error: {:?}, for statement {:?}",
729 e,
730 HexDisplay::from(&statement.hash())
731 );
732 None
733 },
734 Ok(None) => {
735 log::debug!(
736 target: LOG_TARGET,
737 "Keystore is missing key for statement {:?}",
738 HexDisplay::from(&statement.hash())
739 );
740 None
741 },
742 Ok(Some(pair)) => match statement.decrypt_private(&pair.into_inner()) {
743 Ok(r) => r.map(|data| map_f(statement, data)),
744 Err(e) => {
745 log::debug!(
746 target: LOG_TARGET,
747 "Decryption error: {:?}, for statement {:?}",
748 e,
749 HexDisplay::from(&statement.hash())
750 );
751 None
752 },
753 },
754 }
755 } else {
756 None
757 }
758 })
759 }
760}
761
762impl StatementStore for Store {
763 fn statements(&self) -> Result<Vec<(Hash, Statement)>> {
765 let index = self.index.read();
766 let mut result = Vec::with_capacity(index.entries.len());
767 for h in self.index.read().entries.keys() {
768 let encoded = self.db.get(col::STATEMENTS, h).map_err(|e| Error::Db(e.to_string()))?;
769 if let Some(encoded) = encoded {
770 if let Ok(statement) = Statement::decode(&mut encoded.as_slice()) {
771 let hash = statement.hash();
772 result.push((hash, statement));
773 }
774 }
775 }
776 Ok(result)
777 }
778
779 fn statement(&self, hash: &Hash) -> Result<Option<Statement>> {
781 Ok(
782 match self
783 .db
784 .get(col::STATEMENTS, hash.as_slice())
785 .map_err(|e| Error::Db(e.to_string()))?
786 {
787 Some(entry) => {
788 log::trace!(
789 target: LOG_TARGET,
790 "Queried statement {:?}",
791 HexDisplay::from(hash)
792 );
793 Some(
794 Statement::decode(&mut entry.as_slice())
795 .map_err(|e| Error::Decode(e.to_string()))?,
796 )
797 },
798 None => {
799 log::trace!(
800 target: LOG_TARGET,
801 "Queried missing statement {:?}",
802 HexDisplay::from(hash)
803 );
804 None
805 },
806 },
807 )
808 }
809
810 fn broadcasts(&self, match_all_topics: &[Topic]) -> Result<Vec<Vec<u8>>> {
813 self.collect_statements(None, match_all_topics, |statement| statement.into_data())
814 }
815
816 fn posted(&self, match_all_topics: &[Topic], dest: [u8; 32]) -> Result<Vec<Vec<u8>>> {
820 self.collect_statements(Some(dest), match_all_topics, |statement| statement.into_data())
821 }
822
823 fn posted_clear(&self, match_all_topics: &[Topic], dest: [u8; 32]) -> Result<Vec<Vec<u8>>> {
826 self.posted_clear_inner(match_all_topics, dest, |_statement, data| data)
827 }
828
829 fn broadcasts_stmt(&self, match_all_topics: &[Topic]) -> Result<Vec<Vec<u8>>> {
832 self.collect_statements(None, match_all_topics, |statement| Some(statement.encode()))
833 }
834
835 fn posted_stmt(&self, match_all_topics: &[Topic], dest: [u8; 32]) -> Result<Vec<Vec<u8>>> {
839 self.collect_statements(Some(dest), match_all_topics, |statement| Some(statement.encode()))
840 }
841
842 fn posted_clear_stmt(
845 &self,
846 match_all_topics: &[Topic],
847 dest: [u8; 32],
848 ) -> Result<Vec<Vec<u8>>> {
849 self.posted_clear_inner(match_all_topics, dest, |statement, data| {
850 let mut res = Vec::with_capacity(statement.size_hint() + data.len());
851 statement.encode_to(&mut res);
852 res.extend_from_slice(&data);
853 res
854 })
855 }
856
857 fn submit(&self, statement: Statement, source: StatementSource) -> SubmitResult {
859 let hash = statement.hash();
860 match self.index.read().query(&hash) {
861 IndexQuery::Expired =>
862 if !source.can_be_resubmitted() {
863 return SubmitResult::KnownExpired
864 },
865 IndexQuery::Exists =>
866 if !source.can_be_resubmitted() {
867 return SubmitResult::Known
868 },
869 IndexQuery::Unknown => {},
870 }
871
872 let Some(account_id) = statement.account_id() else {
873 log::debug!(
874 target: LOG_TARGET,
875 "Statement validation failed: Missing proof ({:?})",
876 HexDisplay::from(&hash),
877 );
878 self.metrics.report(|metrics| metrics.validations_invalid.inc());
879 return SubmitResult::Bad("No statement proof")
880 };
881
882 let at_block = if let Some(Proof::OnChain { block_hash, .. }) = statement.proof() {
884 Some(*block_hash)
885 } else {
886 None
887 };
888 let validation_result = (self.validate_fn)(at_block, source, statement.clone());
889 let validation = match validation_result {
890 Ok(validation) => validation,
891 Err(InvalidStatement::BadProof) => {
892 log::debug!(
893 target: LOG_TARGET,
894 "Statement validation failed: BadProof, {:?}",
895 HexDisplay::from(&hash),
896 );
897 self.metrics.report(|metrics| metrics.validations_invalid.inc());
898 return SubmitResult::Bad("Bad statement proof")
899 },
900 Err(InvalidStatement::NoProof) => {
901 log::debug!(
902 target: LOG_TARGET,
903 "Statement validation failed: NoProof, {:?}",
904 HexDisplay::from(&hash),
905 );
906 self.metrics.report(|metrics| metrics.validations_invalid.inc());
907 return SubmitResult::Bad("Missing statement proof")
908 },
909 Err(InvalidStatement::InternalError) =>
910 return SubmitResult::InternalError(Error::Runtime),
911 };
912
913 let current_time = self.timestamp();
914 let mut commit = Vec::new();
915 {
916 let mut index = self.index.write();
917
918 let evicted =
919 match index.insert(hash, &statement, &account_id, &validation, current_time) {
920 MaybeInserted::Ignored => return SubmitResult::Ignored,
921 MaybeInserted::Inserted(evicted) => evicted,
922 };
923
924 commit.push((col::STATEMENTS, hash.to_vec(), Some(statement.encode())));
925 for hash in evicted {
926 commit.push((col::STATEMENTS, hash.to_vec(), None));
927 commit.push((col::EXPIRED, hash.to_vec(), Some((hash, current_time).encode())));
928 }
929 if let Err(e) = self.db.commit(commit) {
930 log::debug!(
931 target: LOG_TARGET,
932 "Statement validation failed: database error {}, {:?}",
933 e,
934 statement
935 );
936 return SubmitResult::InternalError(Error::Db(e.to_string()))
937 }
938 } self.metrics.report(|metrics| metrics.submitted_statements.inc());
940 let network_priority = NetworkPriority::High;
941 log::trace!(target: LOG_TARGET, "Statement submitted: {:?}", HexDisplay::from(&hash));
942 SubmitResult::New(network_priority)
943 }
944
945 fn remove(&self, hash: &Hash) -> Result<()> {
947 let current_time = self.timestamp();
948 {
949 let mut index = self.index.write();
950 if index.make_expired(hash, current_time) {
951 let commit = [
952 (col::STATEMENTS, hash.to_vec(), None),
953 (col::EXPIRED, hash.to_vec(), Some((hash, current_time).encode())),
954 ];
955 if let Err(e) = self.db.commit(commit) {
956 log::debug!(
957 target: LOG_TARGET,
958 "Error removing statement: database error {}, {:?}",
959 e,
960 HexDisplay::from(hash),
961 );
962 return Err(Error::Db(e.to_string()))
963 }
964 }
965 }
966 Ok(())
967 }
968}
969
970#[cfg(test)]
971mod tests {
972 use crate::Store;
973 use sc_keystore::Keystore;
974 use sp_core::{Decode, Encode, Pair};
975 use sp_statement_store::{
976 runtime_api::{InvalidStatement, ValidStatement, ValidateStatement},
977 AccountId, Channel, DecryptionKey, NetworkPriority, Proof, SignatureVerificationResult,
978 Statement, StatementSource, StatementStore, SubmitResult, Topic,
979 };
980
981 type Extrinsic = sp_runtime::OpaqueExtrinsic;
982 type Hash = sp_core::H256;
983 type Hashing = sp_runtime::traits::BlakeTwo256;
984 type BlockNumber = u64;
985 type Header = sp_runtime::generic::Header<BlockNumber, Hashing>;
986 type Block = sp_runtime::generic::Block<Header, Extrinsic>;
987
988 const CORRECT_BLOCK_HASH: [u8; 32] = [1u8; 32];
989
990 #[derive(Clone)]
991 pub(crate) struct TestClient;
992
993 pub(crate) struct RuntimeApi {
994 _inner: TestClient,
995 }
996
997 impl sp_api::ProvideRuntimeApi<Block> for TestClient {
998 type Api = RuntimeApi;
999 fn runtime_api(&self) -> sp_api::ApiRef<Self::Api> {
1000 RuntimeApi { _inner: self.clone() }.into()
1001 }
1002 }
1003
1004 sp_api::mock_impl_runtime_apis! {
1005 impl ValidateStatement<Block> for RuntimeApi {
1006 fn validate_statement(
1007 _source: StatementSource,
1008 statement: Statement,
1009 ) -> std::result::Result<ValidStatement, InvalidStatement> {
1010 use crate::tests::account;
1011 match statement.verify_signature() {
1012 SignatureVerificationResult::Valid(_) => Ok(ValidStatement{max_count: 100, max_size: 1000}),
1013 SignatureVerificationResult::Invalid => Err(InvalidStatement::BadProof),
1014 SignatureVerificationResult::NoSignature => {
1015 if let Some(Proof::OnChain { block_hash, .. }) = statement.proof() {
1016 if block_hash == &CORRECT_BLOCK_HASH {
1017 let (max_count, max_size) = match statement.account_id() {
1018 Some(a) if a == account(1) => (1, 1000),
1019 Some(a) if a == account(2) => (2, 1000),
1020 Some(a) if a == account(3) => (3, 1000),
1021 Some(a) if a == account(4) => (4, 1000),
1022 _ => (2, 2000),
1023 };
1024 Ok(ValidStatement{ max_count, max_size })
1025 } else {
1026 Err(InvalidStatement::BadProof)
1027 }
1028 } else {
1029 Err(InvalidStatement::BadProof)
1030 }
1031 }
1032 }
1033 }
1034 }
1035 }
1036
1037 impl sp_blockchain::HeaderBackend<Block> for TestClient {
1038 fn header(&self, _hash: Hash) -> sp_blockchain::Result<Option<Header>> {
1039 unimplemented!()
1040 }
1041 fn info(&self) -> sp_blockchain::Info<Block> {
1042 sp_blockchain::Info {
1043 best_hash: CORRECT_BLOCK_HASH.into(),
1044 best_number: 0,
1045 genesis_hash: Default::default(),
1046 finalized_hash: CORRECT_BLOCK_HASH.into(),
1047 finalized_number: 1,
1048 finalized_state: None,
1049 number_leaves: 0,
1050 block_gap: None,
1051 }
1052 }
1053 fn status(&self, _hash: Hash) -> sp_blockchain::Result<sp_blockchain::BlockStatus> {
1054 unimplemented!()
1055 }
1056 fn number(&self, _hash: Hash) -> sp_blockchain::Result<Option<BlockNumber>> {
1057 unimplemented!()
1058 }
1059 fn hash(&self, _number: BlockNumber) -> sp_blockchain::Result<Option<Hash>> {
1060 unimplemented!()
1061 }
1062 }
1063
1064 fn test_store() -> (Store, tempfile::TempDir) {
1065 sp_tracing::init_for_tests();
1066 let temp_dir = tempfile::Builder::new().tempdir().expect("Error creating test dir");
1067
1068 let client = std::sync::Arc::new(TestClient);
1069 let mut path: std::path::PathBuf = temp_dir.path().into();
1070 path.push("db");
1071 let keystore = std::sync::Arc::new(sc_keystore::LocalKeystore::in_memory());
1072 let store = Store::new(&path, Default::default(), client, keystore, None).unwrap();
1073 (store, temp_dir) }
1075
1076 fn signed_statement(data: u8) -> Statement {
1077 signed_statement_with_topics(data, &[], None)
1078 }
1079
1080 fn signed_statement_with_topics(
1081 data: u8,
1082 topics: &[Topic],
1083 dec_key: Option<DecryptionKey>,
1084 ) -> Statement {
1085 let mut statement = Statement::new();
1086 statement.set_plain_data(vec![data]);
1087 for i in 0..topics.len() {
1088 statement.set_topic(i, topics[i]);
1089 }
1090 if let Some(key) = dec_key {
1091 statement.set_decryption_key(key);
1092 }
1093 let kp = sp_core::ed25519::Pair::from_string("//Alice", None).unwrap();
1094 statement.sign_ed25519_private(&kp);
1095 statement
1096 }
1097
1098 fn topic(data: u64) -> Topic {
1099 let mut topic: Topic = Default::default();
1100 topic[0..8].copy_from_slice(&data.to_le_bytes());
1101 topic
1102 }
1103
1104 fn dec_key(data: u64) -> DecryptionKey {
1105 let mut dec_key: DecryptionKey = Default::default();
1106 dec_key[0..8].copy_from_slice(&data.to_le_bytes());
1107 dec_key
1108 }
1109
1110 fn account(id: u64) -> AccountId {
1111 let mut account: AccountId = Default::default();
1112 account[0..8].copy_from_slice(&id.to_le_bytes());
1113 account
1114 }
1115
1116 fn channel(id: u64) -> Channel {
1117 let mut channel: Channel = Default::default();
1118 channel[0..8].copy_from_slice(&id.to_le_bytes());
1119 channel
1120 }
1121
1122 fn statement(account_id: u64, priority: u32, c: Option<u64>, data_len: usize) -> Statement {
1123 let mut statement = Statement::new();
1124 let mut data = Vec::new();
1125 data.resize(data_len, 0);
1126 statement.set_plain_data(data);
1127 statement.set_priority(priority);
1128 if let Some(c) = c {
1129 statement.set_channel(channel(c));
1130 }
1131 statement.set_proof(Proof::OnChain {
1132 block_hash: CORRECT_BLOCK_HASH,
1133 who: account(account_id),
1134 event_index: 0,
1135 });
1136 statement
1137 }
1138
1139 #[test]
1140 fn submit_one() {
1141 let (store, _temp) = test_store();
1142 let statement0 = signed_statement(0);
1143 assert_eq!(
1144 store.submit(statement0, StatementSource::Network),
1145 SubmitResult::New(NetworkPriority::High)
1146 );
1147 let unsigned = statement(0, 1, None, 0);
1148 assert_eq!(
1149 store.submit(unsigned, StatementSource::Network),
1150 SubmitResult::New(NetworkPriority::High)
1151 );
1152 }
1153
1154 #[test]
1155 fn save_and_load_statements() {
1156 let (store, temp) = test_store();
1157 let statement0 = signed_statement(0);
1158 let statement1 = signed_statement(1);
1159 let statement2 = signed_statement(2);
1160 assert_eq!(
1161 store.submit(statement0.clone(), StatementSource::Network),
1162 SubmitResult::New(NetworkPriority::High)
1163 );
1164 assert_eq!(
1165 store.submit(statement1.clone(), StatementSource::Network),
1166 SubmitResult::New(NetworkPriority::High)
1167 );
1168 assert_eq!(
1169 store.submit(statement2.clone(), StatementSource::Network),
1170 SubmitResult::New(NetworkPriority::High)
1171 );
1172 assert_eq!(store.statements().unwrap().len(), 3);
1173 assert_eq!(store.broadcasts(&[]).unwrap().len(), 3);
1174 assert_eq!(store.statement(&statement1.hash()).unwrap(), Some(statement1.clone()));
1175 let keystore = store.keystore.clone();
1176 drop(store);
1177
1178 let client = std::sync::Arc::new(TestClient);
1179 let mut path: std::path::PathBuf = temp.path().into();
1180 path.push("db");
1181 let store = Store::new(&path, Default::default(), client, keystore, None).unwrap();
1182 assert_eq!(store.statements().unwrap().len(), 3);
1183 assert_eq!(store.broadcasts(&[]).unwrap().len(), 3);
1184 assert_eq!(store.statement(&statement1.hash()).unwrap(), Some(statement1));
1185 }
1186
1187 #[test]
1188 fn search_by_topic_and_key() {
1189 let (store, _temp) = test_store();
1190 let statement0 = signed_statement(0);
1191 let statement1 = signed_statement_with_topics(1, &[topic(0)], None);
1192 let statement2 = signed_statement_with_topics(2, &[topic(0), topic(1)], Some(dec_key(2)));
1193 let statement3 = signed_statement_with_topics(3, &[topic(0), topic(1), topic(2)], None);
1194 let statement4 =
1195 signed_statement_with_topics(4, &[topic(0), topic(42), topic(2), topic(3)], None);
1196 let statements = vec![statement0, statement1, statement2, statement3, statement4];
1197 for s in &statements {
1198 store.submit(s.clone(), StatementSource::Network);
1199 }
1200
1201 let assert_topics = |topics: &[u64], key: Option<u64>, expected: &[u8]| {
1202 let key = key.map(dec_key);
1203 let topics: Vec<_> = topics.iter().map(|t| topic(*t)).collect();
1204 let mut got_vals: Vec<_> = if let Some(key) = key {
1205 store.posted(&topics, key).unwrap().into_iter().map(|d| d[0]).collect()
1206 } else {
1207 store.broadcasts(&topics).unwrap().into_iter().map(|d| d[0]).collect()
1208 };
1209 got_vals.sort();
1210 assert_eq!(expected.to_vec(), got_vals);
1211 };
1212
1213 assert_topics(&[], None, &[0, 1, 3, 4]);
1214 assert_topics(&[], Some(2), &[2]);
1215 assert_topics(&[0], None, &[1, 3, 4]);
1216 assert_topics(&[1], None, &[3]);
1217 assert_topics(&[2], None, &[3, 4]);
1218 assert_topics(&[3], None, &[4]);
1219 assert_topics(&[42], None, &[4]);
1220
1221 assert_topics(&[0, 1], None, &[3]);
1222 assert_topics(&[0, 1], Some(2), &[2]);
1223 assert_topics(&[0, 1, 99], Some(2), &[]);
1224 assert_topics(&[1, 2], None, &[3]);
1225 assert_topics(&[99], None, &[]);
1226 assert_topics(&[0, 99], None, &[]);
1227 assert_topics(&[0, 1, 2, 3, 42], None, &[]);
1228 }
1229
1230 #[test]
1231 fn constraints() {
1232 let (store, _temp) = test_store();
1233
1234 store.index.write().options.max_total_size = 3000;
1235 let source = StatementSource::Network;
1236 let ok = SubmitResult::New(NetworkPriority::High);
1237 let ignored = SubmitResult::Ignored;
1238
1239 assert_eq!(store.submit(statement(1, 1, Some(1), 2000), source), ignored);
1243 assert_eq!(store.submit(statement(1, 1, Some(1), 500), source), ok);
1244 assert_eq!(store.submit(statement(1, 1, Some(1), 200), source), ignored);
1246 assert_eq!(store.submit(statement(1, 2, Some(1), 600), source), ok);
1247 assert_eq!(store.submit(statement(1, 1, Some(2), 100), source), ignored);
1250 assert_eq!(store.index.read().expired.len(), 1);
1251
1252 assert_eq!(store.submit(statement(2, 1, None, 500), source), ok);
1255 assert_eq!(store.submit(statement(2, 2, None, 100), source), ok);
1256 assert_eq!(store.submit(statement(2, 3, None, 500), source), ok);
1258 assert_eq!(store.index.read().expired.len(), 2);
1259 assert_eq!(store.submit(statement(2, 4, None, 1000), source), ok);
1261 assert_eq!(store.index.read().expired.len(), 4);
1262
1263 assert_eq!(store.submit(statement(3, 2, Some(1), 300), source), ok);
1266 assert_eq!(store.submit(statement(3, 3, Some(2), 300), source), ok);
1267 assert_eq!(store.submit(statement(3, 4, Some(3), 300), source), ok);
1268 assert_eq!(store.submit(statement(3, 5, None, 500), source), ok);
1270 assert_eq!(store.index.read().expired.len(), 6);
1271
1272 assert_eq!(store.index.read().total_size, 2400);
1273 assert_eq!(store.index.read().entries.len(), 4);
1274
1275 assert_eq!(store.submit(statement(1, 1, None, 700), source), ignored);
1277 store.index.write().options.max_total_statements = 4;
1279 assert_eq!(store.submit(statement(1, 1, None, 100), source), ignored);
1280
1281 let mut expected_statements = vec![
1282 statement(1, 2, Some(1), 600).hash(),
1283 statement(2, 4, None, 1000).hash(),
1284 statement(3, 4, Some(3), 300).hash(),
1285 statement(3, 5, None, 500).hash(),
1286 ];
1287 expected_statements.sort();
1288 let mut statements: Vec<_> =
1289 store.statements().unwrap().into_iter().map(|(hash, _)| hash).collect();
1290 statements.sort();
1291 assert_eq!(expected_statements, statements);
1292 }
1293
1294 #[test]
1295 fn expired_statements_are_purged() {
1296 use super::DEFAULT_PURGE_AFTER_SEC;
1297 let (mut store, temp) = test_store();
1298 let mut statement = statement(1, 1, Some(3), 100);
1299 store.set_time(0);
1300 statement.set_topic(0, topic(4));
1301 store.submit(statement.clone(), StatementSource::Network);
1302 assert_eq!(store.index.read().entries.len(), 1);
1303 store.remove(&statement.hash()).unwrap();
1304 assert_eq!(store.index.read().entries.len(), 0);
1305 assert_eq!(store.index.read().accounts.len(), 0);
1306 store.set_time(DEFAULT_PURGE_AFTER_SEC + 1);
1307 store.maintain();
1308 assert_eq!(store.index.read().expired.len(), 0);
1309 let keystore = store.keystore.clone();
1310 drop(store);
1311
1312 let client = std::sync::Arc::new(TestClient);
1313 let mut path: std::path::PathBuf = temp.path().into();
1314 path.push("db");
1315 let store = Store::new(&path, Default::default(), client, keystore, None).unwrap();
1316 assert_eq!(store.statements().unwrap().len(), 0);
1317 assert_eq!(store.index.read().expired.len(), 0);
1318 }
1319
1320 #[test]
1321 fn posted_clear_decrypts() {
1322 let (store, _temp) = test_store();
1323 let public = store
1324 .keystore
1325 .ed25519_generate_new(sp_core::crypto::key_types::STATEMENT, None)
1326 .unwrap();
1327 let statement1 = statement(1, 1, None, 100);
1328 let mut statement2 = statement(1, 2, None, 0);
1329 let plain = b"The most valuable secret".to_vec();
1330 statement2.encrypt(&plain, &public).unwrap();
1331 store.submit(statement1, StatementSource::Network);
1332 store.submit(statement2, StatementSource::Network);
1333 let posted_clear = store.posted_clear(&[], public.into()).unwrap();
1334 assert_eq!(posted_clear, vec![plain]);
1335 }
1336
1337 #[test]
1338 fn broadcasts_stmt_returns_encoded_statements() {
1339 let (store, _tmp) = test_store();
1340
1341 let s0 = signed_statement_with_topics(0, &[], None);
1343 let s1 = signed_statement_with_topics(1, &[topic(42)], None);
1345 let s2 = signed_statement_with_topics(2, &[topic(42)], Some(dec_key(99)));
1347
1348 for s in [&s0, &s1, &s2] {
1349 store.submit(s.clone(), StatementSource::Network);
1350 }
1351
1352 let mut hashes: Vec<_> = store
1354 .broadcasts_stmt(&[])
1355 .unwrap()
1356 .into_iter()
1357 .map(|bytes| Statement::decode(&mut &bytes[..]).unwrap().hash())
1358 .collect();
1359 hashes.sort();
1360 let expected_hashes = {
1361 let mut e = vec![s0.hash(), s1.hash()];
1362 e.sort();
1363 e
1364 };
1365 assert_eq!(hashes, expected_hashes);
1366
1367 let got = store.broadcasts_stmt(&[topic(42)]).unwrap();
1369 assert_eq!(got.len(), 1);
1370 let st = Statement::decode(&mut &got[0][..]).unwrap();
1371 assert_eq!(st.hash(), s1.hash());
1372 }
1373
1374 #[test]
1375 fn posted_stmt_returns_encoded_statements_for_dest() {
1376 let (store, _tmp) = test_store();
1377
1378 let public1 = store
1379 .keystore
1380 .ed25519_generate_new(sp_core::crypto::key_types::STATEMENT, None)
1381 .unwrap();
1382 let dest: [u8; 32] = public1.into();
1383
1384 let public2 = store
1385 .keystore
1386 .ed25519_generate_new(sp_core::crypto::key_types::STATEMENT, None)
1387 .unwrap();
1388
1389 let mut s_with_key = statement(1, 1, None, 0);
1391 let plain1 = b"The most valuable secret".to_vec();
1392 s_with_key.encrypt(&plain1, &public1).unwrap();
1393
1394 let mut s_other_key = statement(2, 2, None, 0);
1396 let plain2 = b"The second most valuable secret".to_vec();
1397 s_other_key.encrypt(&plain2, &public2).unwrap();
1398
1399 for s in [&s_with_key, &s_other_key] {
1401 store.submit(s.clone(), StatementSource::Network);
1402 }
1403
1404 let retrieved = store.posted_stmt(&[], dest).unwrap();
1406 assert_eq!(retrieved.len(), 1, "Only one statement has dec_key=dest");
1407
1408 let returned_stmt = Statement::decode(&mut &retrieved[0][..]).unwrap();
1410 assert_eq!(
1411 returned_stmt.hash(),
1412 s_with_key.hash(),
1413 "Returned statement must match s_with_key"
1414 );
1415 }
1416
1417 #[test]
1418 fn posted_clear_stmt_returns_statement_followed_by_plain_data() {
1419 let (store, _tmp) = test_store();
1420
1421 let public1 = store
1422 .keystore
1423 .ed25519_generate_new(sp_core::crypto::key_types::STATEMENT, None)
1424 .unwrap();
1425 let dest: [u8; 32] = public1.into();
1426
1427 let public2 = store
1428 .keystore
1429 .ed25519_generate_new(sp_core::crypto::key_types::STATEMENT, None)
1430 .unwrap();
1431
1432 let mut s_with_key = statement(1, 1, None, 0);
1434 let plain1 = b"The most valuable secret".to_vec();
1435 s_with_key.encrypt(&plain1, &public1).unwrap();
1436
1437 let mut s_other_key = statement(2, 2, None, 0);
1439 let plain2 = b"The second most valuable secret".to_vec();
1440 s_other_key.encrypt(&plain2, &public2).unwrap();
1441
1442 for s in [&s_with_key, &s_other_key] {
1444 store.submit(s.clone(), StatementSource::Network);
1445 }
1446
1447 let retrieved = store.posted_clear_stmt(&[], dest).unwrap();
1449 assert_eq!(retrieved.len(), 1, "Only one statement has dec_key=dest");
1450
1451 let encoded_stmt = s_with_key.encode();
1453 let stmt_len = encoded_stmt.len();
1454
1455 assert_eq!(&retrieved[0][..stmt_len], &encoded_stmt[..]);
1457
1458 let trailing = &retrieved[0][stmt_len..];
1460 assert_eq!(trailing, &plain1[..]);
1461 }
1462
1463 #[test]
1464 fn posted_clear_returns_plain_data_for_dest_and_topics() {
1465 let (store, _tmp) = test_store();
1466
1467 let public_dest = store
1469 .keystore
1470 .ed25519_generate_new(sp_core::crypto::key_types::STATEMENT, None)
1471 .unwrap();
1472 let dest: [u8; 32] = public_dest.into();
1473
1474 let public_other = store
1475 .keystore
1476 .ed25519_generate_new(sp_core::crypto::key_types::STATEMENT, None)
1477 .unwrap();
1478
1479 let mut s_good = statement(1, 1, None, 0);
1481 let plaintext_good = b"The most valuable secret".to_vec();
1482 s_good.encrypt(&plaintext_good, &public_dest).unwrap();
1483 s_good.set_topic(0, topic(42));
1484
1485 let mut s_wrong_topic = statement(2, 2, None, 0);
1487 s_wrong_topic.encrypt(b"Wrong topic", &public_dest).unwrap();
1488 s_wrong_topic.set_topic(0, topic(99));
1489
1490 let mut s_other_dest = statement(3, 3, None, 0);
1492 s_other_dest.encrypt(b"Other dest", &public_other).unwrap();
1493 s_other_dest.set_topic(0, topic(42));
1494
1495 for s in [&s_good, &s_wrong_topic, &s_other_dest] {
1497 store.submit(s.clone(), StatementSource::Network);
1498 }
1499
1500 let retrieved = store.posted_clear(&[topic(42)], dest).unwrap();
1502
1503 assert_eq!(retrieved, vec![plaintext_good]);
1505 }
1506}