1use anyhow::{Context, bail};
8use consensus_core::{
9 CommitAPI, CommitIndex, CommitRange,
10 storage::{Store as ConsensusStore, rocksdb_store::RocksDBStore},
11};
12use std::sync::Arc;
13use sui_core::{
14 authority::authority_store_tables::AuthorityPerpetualTables, checkpoints::CheckpointStore,
15 epoch::committee_store::CommitteeStore,
16};
17use sui_types::{
18 base_types::EpochId,
19 digests::{CheckpointContentsDigest, CheckpointDigest, TransactionDigest},
20 messages_checkpoint::{CheckpointContents, CheckpointSequenceNumber, VerifiedCheckpoint},
21};
22
23use crate::db_shell::{
24 backend::{Backend, DirEntry},
25 vfs::VfsPath,
26};
27
28pub struct DirectBackend {
29 pub checkpoint_store: Arc<CheckpointStore>,
30 pub committee_store: Arc<CommitteeStore>,
31 pub authority_tables: Arc<AuthorityPerpetualTables>,
32 pub consensus_store: Option<Arc<RocksDBStore>>,
33}
34
35impl DirectBackend {
36 fn list_checkpoints_from(
37 &self,
38 start: Option<CheckpointSequenceNumber>,
39 limit: usize,
40 ) -> anyhow::Result<Vec<DirEntry>> {
41 Ok(self
42 .checkpoint_store
43 .list_checkpoints_from_seq(start, limit)?
44 .into_iter()
45 .map(|(seq, _)| DirEntry {
46 name: seq.to_string(),
47 is_dir: true,
48 })
49 .collect())
50 }
51
52 fn list_digest_dirs(
53 &self,
54 start: Option<CheckpointDigest>,
55 limit: usize,
56 ) -> anyhow::Result<Vec<DirEntry>> {
57 Ok(self
58 .checkpoint_store
59 .list_checkpoint_digests(start, limit)?
60 .into_iter()
61 .map(|d| DirEntry {
62 name: d.to_string(),
63 is_dir: true,
64 })
65 .collect())
66 }
67
68 fn list_contents_entries(
69 &self,
70 start: Option<CheckpointContentsDigest>,
71 limit: usize,
72 ) -> anyhow::Result<Vec<DirEntry>> {
73 Ok(self
74 .checkpoint_store
75 .list_checkpoint_contents_digests(start, limit)?
76 .into_iter()
77 .map(|d| DirEntry {
78 name: d.to_string(),
79 is_dir: false,
80 })
81 .collect())
82 }
83
84 fn list_epoch_checkpoint_dirs(
85 &self,
86 epoch: EpochId,
87 start: Option<CheckpointSequenceNumber>,
88 limit: usize,
89 ) -> anyhow::Result<Vec<DirEntry>> {
90 Ok(self
91 .checkpoint_store
92 .list_epoch_checkpoints(epoch, start, limit)?
93 .into_iter()
94 .map(|(seq, _)| DirEntry {
95 name: seq.to_string(),
96 is_dir: false,
97 })
98 .collect())
99 }
100
101 fn list_transactions_from(
102 &self,
103 start: Option<TransactionDigest>,
104 limit: usize,
105 ) -> anyhow::Result<Vec<DirEntry>> {
106 let tx_digests = self.authority_tables.list_transactions_from(start, limit)?;
107 let mut entries = Vec::with_capacity(tx_digests.len() * 2);
108 for digest in &tx_digests {
109 entries.push(DirEntry {
110 name: digest.to_string(),
111 is_dir: false,
112 });
113 if let Ok(Some(fx_digest)) = self.authority_tables.get_executed_effects_digest(digest) {
114 entries.push(DirEntry {
115 name: format!("{digest}.fx-{fx_digest}"),
116 is_dir: false,
117 });
118 }
119 }
120 Ok(entries)
121 }
122
123 fn list_consensus_commits(
124 &self,
125 start: Option<CommitIndex>,
126 limit: usize,
127 ) -> anyhow::Result<Vec<DirEntry>> {
128 let cs = self.consensus_store.as_ref().ok_or_else(|| {
129 anyhow::anyhow!("consensus store not available; use --consensus-db-path")
130 })?;
131 let start_idx = start.unwrap_or(1);
132 let commits = cs
133 .scan_commits(CommitRange::new(start_idx..=CommitIndex::MAX))
134 .map_err(|e| anyhow::anyhow!("{e}"))?;
135 Ok(commits
136 .into_iter()
137 .take(limit)
138 .map(|c| DirEntry {
139 name: c.index().to_string(),
140 is_dir: true,
141 })
142 .collect())
143 }
144
145 fn get_checkpoint_by_seq(
146 &self,
147 seq: CheckpointSequenceNumber,
148 ) -> anyhow::Result<VerifiedCheckpoint> {
149 self.checkpoint_store
150 .get_checkpoint_by_sequence_number(seq)?
151 .with_context(|| format!("checkpoint {seq} not found"))
152 }
153
154 fn get_checkpoint_by_digest(
155 &self,
156 digest: &CheckpointDigest,
157 ) -> anyhow::Result<VerifiedCheckpoint> {
158 self.checkpoint_store
159 .get_checkpoint_by_digest(digest)?
160 .with_context(|| format!("checkpoint {digest} not found"))
161 }
162
163 fn get_checkpoint_contents(
164 &self,
165 cp: &VerifiedCheckpoint,
166 ) -> anyhow::Result<CheckpointContents> {
167 self.checkpoint_store
168 .get_checkpoint_contents(&cp.content_digest)?
169 .with_context(|| format!("contents for checkpoint {} not found", cp.sequence_number()))
170 }
171
172 fn contents_short_text(&self, contents: &CheckpointContents) -> String {
173 let mut out = String::new();
174 for ed in contents.iter() {
175 out.push_str(&format!("{} {}\n", ed.transaction, ed.effects));
176 }
177 out
178 }
179
180 fn contents_short_json(
181 &self,
182 contents: &CheckpointContents,
183 ) -> anyhow::Result<serde_json::Value> {
184 let pairs: Vec<_> = contents
185 .iter()
186 .map(|ed| {
187 serde_json::json!({
188 "transaction": ed.transaction.to_string(),
189 "effects": ed.effects.to_string(),
190 })
191 })
192 .collect();
193 Ok(serde_json::Value::Array(pairs))
194 }
195
196 fn summary_json(&self, cp: &VerifiedCheckpoint) -> anyhow::Result<serde_json::Value> {
197 Ok(serde_json::to_value(cp.data())?)
198 }
199
200 fn summary_debug(&self, cp: &VerifiedCheckpoint) -> String {
201 format!("{:#?}", cp.data())
202 }
203
204 fn summary_bcs(&self, cp: &VerifiedCheckpoint) -> anyhow::Result<Vec<u8>> {
205 Ok(bcs::to_bytes(cp.data())?)
206 }
207
208 fn contents_json(&self, contents: &CheckpointContents) -> anyhow::Result<serde_json::Value> {
209 Ok(serde_json::to_value(contents)?)
210 }
211
212 fn contents_debug(&self, contents: &CheckpointContents) -> String {
213 format!("{contents:#?}")
214 }
215
216 fn contents_bcs(&self, contents: &CheckpointContents) -> anyhow::Result<Vec<u8>> {
217 Ok(bcs::to_bytes(contents)?)
218 }
219
220 fn commit_summary_json(&self, index: CommitIndex) -> anyhow::Result<serde_json::Value> {
221 let cs = self.consensus_store.as_ref().ok_or_else(|| {
222 anyhow::anyhow!("consensus store not available; use --consensus-db-path")
223 })?;
224 let summary =
225 sui_core::consensus_commit_summary::build_consensus_commit_summary(cs, index)?
226 .ok_or_else(|| anyhow::anyhow!("commit {index} not found"))?;
227 let commit = &summary.commit;
228 Ok(serde_json::json!({
229 "index": commit.index(),
230 "timestamp_ms": commit.timestamp_ms(),
231 "leader": commit.leader().to_string(),
232 "previous_digest": commit.previous_digest().to_string(),
233 "block_count": commit.blocks().len(),
234 "transactions": summary.tx_keys,
235 "missing_blocks": summary.missing_blocks,
236 }))
237 }
238
239 fn commit_summary_debug(&self, index: CommitIndex) -> anyhow::Result<String> {
240 let json = self.commit_summary_json(index)?;
241 Ok(format!("{json:#}"))
242 }
243}
244
245impl Backend for DirectBackend {
246 fn ls_children(&self, path: &VfsPath, limit: usize) -> anyhow::Result<Vec<DirEntry>> {
247 match path {
248 VfsPath::Root => Ok(vec![
249 DirEntry {
250 name: "epochs".into(),
251 is_dir: true,
252 },
253 DirEntry {
254 name: "checkpoints".into(),
255 is_dir: true,
256 },
257 DirEntry {
258 name: "checkpoint-contents".into(),
259 is_dir: true,
260 },
261 DirEntry {
262 name: "transactions".into(),
263 is_dir: true,
264 },
265 DirEntry {
266 name: "consensus".into(),
267 is_dir: true,
268 },
269 ]),
270 VfsPath::Epochs => {
271 let epochs = self.committee_store.list_epochs(None, limit)?;
272 Ok(epochs
273 .into_iter()
274 .map(|(id, _)| DirEntry {
275 name: id.to_string(),
276 is_dir: true,
277 })
278 .collect())
279 }
280 VfsPath::Epoch(_) => Ok(vec![
281 DirEntry {
282 name: "first-checkpoint".into(),
283 is_dir: false,
284 },
285 DirEntry {
286 name: "last-checkpoint".into(),
287 is_dir: false,
288 },
289 DirEntry {
290 name: "committee".into(),
291 is_dir: false,
292 },
293 DirEntry {
294 name: "checkpoints".into(),
295 is_dir: true,
296 },
297 ]),
298 VfsPath::EpochCheckpoints(epoch) => {
299 self.list_epoch_checkpoint_dirs(*epoch, None, limit)
300 }
301 VfsPath::CheckpointsRoot => Ok(vec![
302 DirEntry {
303 name: "seq".into(),
304 is_dir: true,
305 },
306 DirEntry {
307 name: "digest".into(),
308 is_dir: true,
309 },
310 ]),
311 VfsPath::CheckpointsSeqRoot => self.list_checkpoints_from(None, limit),
312 VfsPath::CheckpointsBySeq(_) => Ok(vec![
313 DirEntry {
314 name: "summary".into(),
315 is_dir: false,
316 },
317 DirEntry {
318 name: "contents".into(),
319 is_dir: false,
320 },
321 DirEntry {
322 name: "contents-short".into(),
323 is_dir: false,
324 },
325 ]),
326 VfsPath::CheckpointsDigestRoot => self.list_digest_dirs(None, limit),
327 VfsPath::CheckpointsByDigest(_) => Ok(vec![
328 DirEntry {
329 name: "summary".into(),
330 is_dir: false,
331 },
332 DirEntry {
333 name: "contents".into(),
334 is_dir: false,
335 },
336 DirEntry {
337 name: "contents-short".into(),
338 is_dir: false,
339 },
340 ]),
341 VfsPath::CheckpointContentsRoot => self.list_contents_entries(None, limit),
342 VfsPath::TransactionsRoot => self.list_transactions_from(None, limit),
343 VfsPath::ConsensusRoot => Ok(vec![
344 DirEntry {
345 name: "latest".into(),
346 is_dir: false,
347 },
348 DirEntry {
349 name: "commits".into(),
350 is_dir: true,
351 },
352 ]),
353 VfsPath::ConsensusCommitsRoot => self.list_consensus_commits(None, limit),
354 VfsPath::ConsensusCommitDir(_) => Ok(vec![DirEntry {
355 name: "summary".into(),
356 is_dir: false,
357 }]),
358 _ => bail!("'{}' is not a directory", path),
359 }
360 }
361
362 fn ls_cursor(&self, path: &VfsPath, limit: usize) -> anyhow::Result<Vec<DirEntry>> {
363 match path {
364 VfsPath::CheckpointsBySeq(seq) => self.list_checkpoints_from(Some(*seq), limit),
365 VfsPath::CheckpointsByDigest(d) => self.list_digest_dirs(Some(*d), limit),
366 VfsPath::EpochCheckpointBySeq(epoch, seq) => {
367 self.list_epoch_checkpoint_dirs(*epoch, Some(*seq), limit)
368 }
369 VfsPath::CheckpointContentsEntry(d) => self.list_contents_entries(Some(*d), limit),
370 VfsPath::TransactionEntry(d) => self.list_transactions_from(Some(*d), limit),
371 VfsPath::ConsensusCommitDir(i) => self.list_consensus_commits(Some(*i), limit),
372 _ => self.ls_children(path, limit),
374 }
375 }
376
377 fn read_json(&self, path: &VfsPath) -> anyhow::Result<serde_json::Value> {
378 match path {
379 VfsPath::EpochFirstCheckpoint(epoch) => {
380 let seq = self
381 .checkpoint_store
382 .get_epoch_first_checkpoint_seq(*epoch)?
383 .with_context(|| format!("no data for epoch {epoch}"))?;
384 let cp = self.get_checkpoint_by_seq(seq)?;
385 self.summary_json(&cp)
386 }
387 VfsPath::EpochLastCheckpoint(epoch) => {
388 let cp = self
389 .checkpoint_store
390 .get_epoch_last_checkpoint(*epoch)?
391 .with_context(|| format!("no last checkpoint for epoch {epoch}"))?;
392 self.summary_json(&cp)
393 }
394 VfsPath::EpochCommittee(epoch) => {
395 let committee = self
396 .committee_store
397 .get_committee(epoch)?
398 .with_context(|| format!("no committee for epoch {epoch}"))?;
399 Ok(serde_json::to_value(committee.as_ref())?)
400 }
401 VfsPath::EpochCheckpointBySeq(_epoch, seq) => {
402 let cp = self.get_checkpoint_by_seq(*seq)?;
403 self.summary_json(&cp)
404 }
405 VfsPath::EpochCheckpointByDigest(_epoch, digest) => {
406 let cp = self.get_checkpoint_by_digest(digest)?;
407 self.summary_json(&cp)
408 }
409 VfsPath::CheckpointSeqSummary(seq) => {
410 let cp = self.get_checkpoint_by_seq(*seq)?;
411 self.summary_json(&cp)
412 }
413 VfsPath::CheckpointSeqContents(seq) => {
414 let cp = self.get_checkpoint_by_seq(*seq)?;
415 let contents = self.get_checkpoint_contents(&cp)?;
416 self.contents_json(&contents)
417 }
418 VfsPath::CheckpointSeqContentsShort(seq) => {
419 let cp = self.get_checkpoint_by_seq(*seq)?;
420 let contents = self.get_checkpoint_contents(&cp)?;
421 self.contents_short_json(&contents)
422 }
423 VfsPath::CheckpointDigestSummary(digest) => {
424 let cp = self.get_checkpoint_by_digest(digest)?;
425 self.summary_json(&cp)
426 }
427 VfsPath::CheckpointDigestContents(digest) => {
428 let cp = self.get_checkpoint_by_digest(digest)?;
429 let contents = self.get_checkpoint_contents(&cp)?;
430 self.contents_json(&contents)
431 }
432 VfsPath::CheckpointDigestContentsShort(digest) => {
433 let cp = self.get_checkpoint_by_digest(digest)?;
434 let contents = self.get_checkpoint_contents(&cp)?;
435 self.contents_short_json(&contents)
436 }
437 VfsPath::CheckpointContentsEntry(digest) => {
438 let contents = self
439 .checkpoint_store
440 .get_checkpoint_contents(digest)?
441 .with_context(|| format!("checkpoint contents {digest} not found"))?;
442 self.contents_json(&contents)
443 }
444 VfsPath::TransactionEntry(digest) => {
445 let tx = self
446 .authority_tables
447 .get_transaction(digest)?
448 .with_context(|| format!("transaction {digest} not found"))?;
449 Ok(serde_json::to_value(&tx)?)
450 }
451 VfsPath::TransactionEffectsEntry(tx_digest, fx_digest) => {
452 let effects = self
453 .authority_tables
454 .get_effects_by_digest(fx_digest)?
455 .with_context(|| format!("effects {fx_digest} for tx {tx_digest} not found"))?;
456 Ok(serde_json::to_value(&effects)?)
457 }
458 VfsPath::ConsensusLatest => {
459 let cs = self
460 .consensus_store
461 .as_ref()
462 .ok_or_else(|| anyhow::anyhow!("--consensus-db-path not provided"))?;
463 let commit = cs
464 .read_last_commit()?
465 .ok_or_else(|| anyhow::anyhow!("no commits yet"))?;
466 Ok(serde_json::json!({ "index": commit.index() }))
467 }
468 VfsPath::ConsensusCommitSummary(index) => self.commit_summary_json(*index),
469 _ => bail!("'{}' is not readable", path),
470 }
471 }
472
473 fn read_debug(&self, path: &VfsPath) -> anyhow::Result<String> {
474 match path {
475 VfsPath::EpochFirstCheckpoint(epoch) => {
476 let seq = self
477 .checkpoint_store
478 .get_epoch_first_checkpoint_seq(*epoch)?
479 .with_context(|| format!("no data for epoch {epoch}"))?;
480 let cp = self.get_checkpoint_by_seq(seq)?;
481 Ok(self.summary_debug(&cp))
482 }
483 VfsPath::EpochLastCheckpoint(epoch) => {
484 let cp = self
485 .checkpoint_store
486 .get_epoch_last_checkpoint(*epoch)?
487 .with_context(|| format!("no last checkpoint for epoch {epoch}"))?;
488 Ok(self.summary_debug(&cp))
489 }
490 VfsPath::EpochCommittee(epoch) => {
491 let committee = self
492 .committee_store
493 .get_committee(epoch)?
494 .with_context(|| format!("no committee for epoch {epoch}"))?;
495 Ok(format!("{:#?}", committee.as_ref()))
496 }
497 VfsPath::EpochCheckpointBySeq(_epoch, seq) => {
498 let cp = self.get_checkpoint_by_seq(*seq)?;
499 Ok(self.summary_debug(&cp))
500 }
501 VfsPath::EpochCheckpointByDigest(_epoch, digest) => {
502 let cp = self.get_checkpoint_by_digest(digest)?;
503 Ok(self.summary_debug(&cp))
504 }
505 VfsPath::CheckpointSeqSummary(seq) => {
506 let cp = self.get_checkpoint_by_seq(*seq)?;
507 Ok(self.summary_debug(&cp))
508 }
509 VfsPath::CheckpointSeqContents(seq) => {
510 let cp = self.get_checkpoint_by_seq(*seq)?;
511 let contents = self.get_checkpoint_contents(&cp)?;
512 Ok(self.contents_debug(&contents))
513 }
514 VfsPath::CheckpointSeqContentsShort(seq) => {
515 let cp = self.get_checkpoint_by_seq(*seq)?;
516 let contents = self.get_checkpoint_contents(&cp)?;
517 Ok(self.contents_short_text(&contents))
518 }
519 VfsPath::CheckpointDigestSummary(digest) => {
520 let cp = self.get_checkpoint_by_digest(digest)?;
521 Ok(self.summary_debug(&cp))
522 }
523 VfsPath::CheckpointDigestContents(digest) => {
524 let cp = self.get_checkpoint_by_digest(digest)?;
525 let contents = self.get_checkpoint_contents(&cp)?;
526 Ok(self.contents_debug(&contents))
527 }
528 VfsPath::CheckpointDigestContentsShort(digest) => {
529 let cp = self.get_checkpoint_by_digest(digest)?;
530 let contents = self.get_checkpoint_contents(&cp)?;
531 Ok(self.contents_short_text(&contents))
532 }
533 VfsPath::CheckpointContentsEntry(digest) => {
534 let contents = self
535 .checkpoint_store
536 .get_checkpoint_contents(digest)?
537 .with_context(|| format!("checkpoint contents {digest} not found"))?;
538 Ok(self.contents_debug(&contents))
539 }
540 VfsPath::TransactionEntry(digest) => {
541 let tx = self
542 .authority_tables
543 .get_transaction(digest)?
544 .with_context(|| format!("transaction {digest} not found"))?;
545 Ok(format!("{tx:#?}"))
546 }
547 VfsPath::TransactionEffectsEntry(tx_digest, fx_digest) => {
548 let effects = self
549 .authority_tables
550 .get_effects_by_digest(fx_digest)?
551 .with_context(|| format!("effects {fx_digest} for tx {tx_digest} not found"))?;
552 Ok(format!("{effects:#?}"))
553 }
554 VfsPath::ConsensusLatest => {
555 let cs = self
556 .consensus_store
557 .as_ref()
558 .ok_or_else(|| anyhow::anyhow!("--consensus-db-path not provided"))?;
559 let commit = cs
560 .read_last_commit()?
561 .ok_or_else(|| anyhow::anyhow!("no commits yet"))?;
562 Ok(commit.index().to_string())
563 }
564 VfsPath::ConsensusCommitSummary(index) => self.commit_summary_debug(*index),
565 _ => bail!("'{}' is not readable", path),
566 }
567 }
568
569 fn read_bcs(&self, path: &VfsPath) -> anyhow::Result<Vec<u8>> {
570 match path {
571 VfsPath::EpochFirstCheckpoint(epoch) => {
572 let seq = self
573 .checkpoint_store
574 .get_epoch_first_checkpoint_seq(*epoch)?
575 .with_context(|| format!("no data for epoch {epoch}"))?;
576 let cp = self.get_checkpoint_by_seq(seq)?;
577 self.summary_bcs(&cp)
578 }
579 VfsPath::EpochLastCheckpoint(epoch) => {
580 let cp = self
581 .checkpoint_store
582 .get_epoch_last_checkpoint(*epoch)?
583 .with_context(|| format!("no last checkpoint for epoch {epoch}"))?;
584 self.summary_bcs(&cp)
585 }
586 VfsPath::EpochCommittee(epoch) => {
587 let committee = self
588 .committee_store
589 .get_committee(epoch)?
590 .with_context(|| format!("no committee for epoch {epoch}"))?;
591 Ok(bcs::to_bytes(committee.as_ref())?)
592 }
593 VfsPath::EpochCheckpointBySeq(_epoch, seq) => {
594 let cp = self.get_checkpoint_by_seq(*seq)?;
595 self.summary_bcs(&cp)
596 }
597 VfsPath::EpochCheckpointByDigest(_epoch, digest) => {
598 let cp = self.get_checkpoint_by_digest(digest)?;
599 self.summary_bcs(&cp)
600 }
601 VfsPath::CheckpointSeqSummary(seq) => {
602 let cp = self.get_checkpoint_by_seq(*seq)?;
603 self.summary_bcs(&cp)
604 }
605 VfsPath::CheckpointSeqContents(seq) => {
606 let cp = self.get_checkpoint_by_seq(*seq)?;
607 let contents = self.get_checkpoint_contents(&cp)?;
608 self.contents_bcs(&contents)
609 }
610 VfsPath::CheckpointDigestSummary(digest) => {
611 let cp = self.get_checkpoint_by_digest(digest)?;
612 self.summary_bcs(&cp)
613 }
614 VfsPath::CheckpointDigestContents(digest) => {
615 let cp = self.get_checkpoint_by_digest(digest)?;
616 let contents = self.get_checkpoint_contents(&cp)?;
617 self.contents_bcs(&contents)
618 }
619 VfsPath::CheckpointContentsEntry(digest) => {
620 let contents = self
621 .checkpoint_store
622 .get_checkpoint_contents(digest)?
623 .with_context(|| format!("checkpoint contents {digest} not found"))?;
624 self.contents_bcs(&contents)
625 }
626 VfsPath::TransactionEntry(digest) => {
627 let tx = self
628 .authority_tables
629 .get_transaction(digest)?
630 .with_context(|| format!("transaction {digest} not found"))?;
631 Ok(bcs::to_bytes(&tx)?)
632 }
633 VfsPath::TransactionEffectsEntry(tx_digest, fx_digest) => {
634 let effects = self
635 .authority_tables
636 .get_effects_by_digest(fx_digest)?
637 .with_context(|| format!("effects {fx_digest} for tx {tx_digest} not found"))?;
638 Ok(bcs::to_bytes(&effects)?)
639 }
640 _ => bail!(
641 "'{}' is not readable or bcs not supported for this entry",
642 path
643 ),
644 }
645 }
646
647 fn delete(&self, _path: &VfsPath) -> anyhow::Result<()> {
648 bail!("delete not yet implemented for direct mode")
649 }
650}