referrerpolicy=no-referrer-when-downgrade

sc_statement_store/
lib.rs

1// This file is part of Substrate.
2
3// Copyright (C) Parity Technologies (UK) Ltd.
4// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
5
6// This program is free software: you can redistribute it and/or modify
7// it under the terms of the GNU General Public License as published by
8// the Free Software Foundation, either version 3 of the License, or
9// (at your option) any later version.
10
11// This program is distributed in the hope that it will be useful,
12// but WITHOUT ANY WARRANTY; without even the implied warranty of
13// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14// GNU General Public License for more details.
15
16// You should have received a copy of the GNU General Public License
17// along with this program. If not, see <https://www.gnu.org/licenses/>.
18
19//! Disk-backed statement store.
20//!
21//! This module contains an implementation of `sp_statement_store::StatementStore` which is backed
22//! by a database.
23//!
24//! Constraint management.
25//!
26//! Each time a new statement is inserted into the store, it is first validated with the runtime
27//! Validation function computes `global_priority`, 'max_count' and `max_size` for a statement.
28//! The following constraints are then checked:
29//! * For a given account id, there may be at most `max_count` statements with `max_size` total data
30//!   size. To satisfy this, statements for this account ID are removed from the store starting with
31//!   the lowest priority until a constraint is satisfied.
32//! * There may not be more than `MAX_TOTAL_STATEMENTS` total statements with `MAX_TOTAL_SIZE` size.
33//!   To satisfy this, statements are removed from the store starting with the lowest
34//!   `global_priority` until a constraint is satisfied.
35//!
36//! When a new statement is inserted that would not satisfy constraints in the first place, no
37//! statements are deleted and `Ignored` result is returned.
38//! The order in which statements with the same priority are deleted is unspecified.
39//!
40//! Statement expiration.
41//!
42//! Each time a statement is removed from the store (Either evicted by higher priority statement or
43//! explicitly with the `remove` function) the statement is marked as expired. Expired statements
44//! can't be added to the store for `Options::purge_after_sec` seconds. This is to prevent old
45//! statements from being propagated on the network.
46
47#![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; //48h
80const 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	// Statements ordered by priority.
123	by_priority: BTreeMap<PriorityKey, (Option<Channel>, usize)>,
124	// Channel to statement map. Only one statement per channel is allowed.
125	channels: HashMap<Channel, ChannelEntry>,
126	// Sum of all `Data` field sizes.
127	data_size: usize,
128}
129
130/// Store configuration
131pub struct Options {
132	/// Maximum statement allowed in the store. Once this limit is reached lower-priority
133	/// statements may be evicted.
134	max_total_statements: usize,
135	/// Maximum total data size allowed in the store. Once this limit is reached lower-priority
136	/// statements may be evicted.
137	max_total_size: usize,
138	/// Number of seconds for which removed statements won't be allowed to be added back in.
139	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>, // Value is expiration timestamp.
159	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			// Validate against the finalized state.
185			self.client.info().finalized_hash
186		});
187		api.validate_statement(block, source, statement)
188			.map_err(|_| InvalidStatement::InternalError)?
189	}
190}
191
192/// Statement store.
193pub 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	// Used for testing
207	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] = [&empty; 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			// Key does not exist in the index.
281			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				// At least one of the match_all_topics does not exist in the index.
288				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		// Start with the smallest topic set or the key set.
294		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		// Purge previously expired messages.
310		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		// It may happen that we can't delete enough lower priority messages
392		// to satisfy size constraints. We check for that before deleting anything,
393		// taking into account channel message replacement.
394		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						// Trying to replace channel message with lower priority
399						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						// Would replace channel message. Still need to check for size constraints
409						// below.
410						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			// Check if we can evict enough lower priority statements to satisfy constraints
430			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					// Satisfied
435					break
436				}
437				if evicted.contains(&entry.hash) {
438					// Already accounted for above
439					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		// Now check global constraints as well.
456		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	/// Create a new shared store instance. There should only be one per process.
479	/// `path` will be used to open a statement database or create a new one if it does not exist.
480	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		// Perform periodic statement store maintenance
502		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	/// Create a new instance.
519	/// `path` will be used to open a statement database or create a new one if it does not exist.
520	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	/// Create memory index from the data.
582	// This may be moved to a background thread if it slows startup too much.
583	// This function should only be used on startup. There should be no other DB operations when
584	// iterating the index.
585	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						// DB inconsistency
650						log::warn!(
651							target: LOG_TARGET,
652							"Corrupt statement {:?}",
653							HexDisplay::from(hash)
654						);
655					}
656				},
657				None => {
658					// DB inconsistency
659					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	/// Perform periodic store maintenance
672	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	/// Returns `self` as [`StatementStoreExt`].
707	pub fn as_statement_store_ext(self: Arc<Self>) -> StatementStoreExt {
708		StatementStoreExt::new(self)
709	}
710
711	/// Return information of all known statements whose decryption key is identified as
712	/// `dest`. The key must be available to the client.
713	fn posted_clear_inner<R>(
714		&self,
715		match_all_topics: &[Topic],
716		dest: [u8; 32],
717		// Map the statement and the decrypted data to the desired result.
718		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	/// Return all statements.
764	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	/// Returns a statement by hash.
780	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	/// Return the data of all known statements which include all topics and have no `DecryptionKey`
811	/// field.
812	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	/// Return the data of all known statements whose decryption key is identified as `dest` (this
817	/// will generally be the public key or a hash thereof for symmetric ciphers, or a hash of the
818	/// private key for symmetric ciphers).
819	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	/// Return the decrypted data of all known statements whose decryption key is identified as
824	/// `dest`. The key must be available to the client.
825	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	/// Return all known statements which include all topics and have no `DecryptionKey`
830	/// field.
831	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	/// Return all known statements whose decryption key is identified as `dest` (this
836	/// will generally be the public key or a hash thereof for symmetric ciphers, or a hash of the
837	/// private key for symmetric ciphers).
838	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	/// Return the statement and the decrypted data of all known statements whose decryption key is
843	/// identified as `dest`. The key must be available to the client.
844	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	/// Submit a statement to the store. Validates the statement and returns validation result.
858	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		// Validate.
883		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		} // Release index lock
939		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	/// Remove a statement by hash.
946	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) // return order is important. Store must be dropped before TempDir
1074	}
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		// Account 1 (limit = 1 msg, 1000 bytes)
1240
1241		// Oversized statement is not allowed. Limit for account 1 is 1 msg, 1000 bytes
1242		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		// Would not replace channel message with same priority
1245		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		// Submit another message to another channel with lower priority. Should not be allowed
1248		// because msg count limit is 1
1249		assert_eq!(store.submit(statement(1, 1, Some(2), 100), source), ignored);
1250		assert_eq!(store.index.read().expired.len(), 1);
1251
1252		// Account 2 (limit = 2 msg, 1000 bytes)
1253
1254		assert_eq!(store.submit(statement(2, 1, None, 500), source), ok);
1255		assert_eq!(store.submit(statement(2, 2, None, 100), source), ok);
1256		// Should evict priority 1
1257		assert_eq!(store.submit(statement(2, 3, None, 500), source), ok);
1258		assert_eq!(store.index.read().expired.len(), 2);
1259		// Should evict all
1260		assert_eq!(store.submit(statement(2, 4, None, 1000), source), ok);
1261		assert_eq!(store.index.read().expired.len(), 4);
1262
1263		// Account 3 (limit = 3 msg, 1000 bytes)
1264
1265		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		// Should evict 2 and 3
1269		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		// Should be over the global size limit
1276		assert_eq!(store.submit(statement(1, 1, None, 700), source), ignored);
1277		// Should be over the global count limit
1278		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		// no key, no topic
1342		let s0 = signed_statement_with_topics(0, &[], None);
1343		// same, but with a topic = 42
1344		let s1 = signed_statement_with_topics(1, &[topic(42)], None);
1345		// has a decryption key -> must NOT be returned by broadcasts_stmt
1346		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		// no topic filter
1353		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		// filter on topic 42
1368		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		// A statement that does have dec_key = dest
1390		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		// A statement with a different dec_key
1395		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		// Submit them all
1400		for s in [&s_with_key, &s_other_key] {
1401			store.submit(s.clone(), StatementSource::Network);
1402		}
1403
1404		// posted_stmt should only return the one with dec_key = dest
1405		let retrieved = store.posted_stmt(&[], dest).unwrap();
1406		assert_eq!(retrieved.len(), 1, "Only one statement has dec_key=dest");
1407
1408		// Re-decode that returned statement to confirm it is correct
1409		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		// A statement that does have dec_key = dest
1433		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		// A statement with a different dec_key
1438		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		// Submit them all
1443		for s in [&s_with_key, &s_other_key] {
1444			store.submit(s.clone(), StatementSource::Network);
1445		}
1446
1447		// posted_stmt should only return the one with dec_key = dest
1448		let retrieved = store.posted_clear_stmt(&[], dest).unwrap();
1449		assert_eq!(retrieved.len(), 1, "Only one statement has dec_key=dest");
1450
1451		// We expect: [ encoded Statement ] + [ the decrypted bytes ]
1452		let encoded_stmt = s_with_key.encode();
1453		let stmt_len = encoded_stmt.len();
1454
1455		// 1) statement is first
1456		assert_eq!(&retrieved[0][..stmt_len], &encoded_stmt[..]);
1457
1458		// 2) followed by the decrypted payload
1459		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		// prepare two key-pairs
1468		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		// statement that SHOULD be returned (matches dest & topic 42)
1480		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		// statement that should NOT be returned (same dest but different topic)
1486		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		// statement that should NOT be returned (different dest)
1491		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		// submit all
1496		for s in [&s_good, &s_wrong_topic, &s_other_dest] {
1497			store.submit(s.clone(), StatementSource::Network);
1498		}
1499
1500		// call posted_clear with the topic filter and dest
1501		let retrieved = store.posted_clear(&[topic(42)], dest).unwrap();
1502
1503		// exactly one element, equal to the expected plaintext
1504		assert_eq!(retrieved, vec![plaintext_good]);
1505	}
1506}