merge upstream/main

Signed-off-by: William Hankins <[email protected]>

Author: whankinsiv

Makefile

@@ -6,6 +6,8 @@ SHELL := bash
 CARGO := cargo
 PYTHON := python3
 PROCESS_PKG := acropolis_process_omnibus
+LOG_LEVEL ?= info
+
 
 # Test snapshots
 SNAPSHOT_SMALL ?= tests/fixtures/snapshot-small.cbor
@@ -28,6 +30,7 @@ help:
 	@echo "Build & Test:"
 	@echo "  all                      Format, lint, and test"
 	@echo "  build                    Build the omnibus process"
+	@echo "  run 					  Run the omnibus"
 	@echo "  test                     Run all tests"
 	@echo "  fmt                      Run cargo fmt"
 	@echo "  clippy                   Run cargo clippy -D warnings"
@@ -37,9 +40,11 @@ help:
 	@echo ""
 	@echo "Variables:"
 	@echo "  SNAPSHOT=<path>          Path to snapshot file (default: Conway epoch 507)"
+	@echo "  LOG_LEVEL=<level>        Set log level (default: info, options: error, warn, info, debug, trace)"
 	@echo ""
 	@echo "Examples:"
 	@echo "  make snap-test-streaming"
+	@echo "  make run LOG_LEVEL=debug"
 	@echo "  make snap-test-streaming SNAPSHOT=path/to/snapshot.cbor"
 
 all: fmt clippy test
@@ -51,7 +56,7 @@ test:
 	$(CARGO) test
 
 run:
-	$(CARGO) run -p $(PROCESS_PKG)
+	cd processes/omnibus && RUST_LOG=$(LOG_LEVEL) $(CARGO) run --release --bin $(PROCESS_PKG)
 
 fmt:
 	$(CARGO) fmt --all

common/src/address.rs

@@ -148,10 +148,12 @@ pub struct ShelleyAddressPointer {
     serde::Deserialize,
     minicbor::Encode,
     minicbor::Decode,
+    Default,
 )]
 pub enum ShelleyAddressDelegationPart {
     /// No delegation (enterprise addresses)
     #[n(0)]
+    #[default]
     None,
 
     /// Delegation to stake key
@@ -167,12 +169,6 @@ pub enum ShelleyAddressDelegationPart {
     Pointer(#[n(0)] ShelleyAddressPointer),
 }
 
-impl Default for ShelleyAddressDelegationPart {
-    fn default() -> Self {
-        Self::None
-    }
-}
-
 /// A Shelley-era address
 #[derive(
     Debug,
@@ -619,20 +615,15 @@ impl Default for StakeAddress {
 }
 
 /// A Cardano address
-#[derive(Debug, Clone, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
+#[derive(Debug, Clone, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize, Default)]
 pub enum Address {
+    #[default]
     None,
     Byron(ByronAddress),
     Shelley(ShelleyAddress),
     Stake(StakeAddress),
 }
 
-impl Default for Address {
-    fn default() -> Self {
-        Self::None
-    }
-}
-
 impl Address {
     /// Get the stake pointer if there is one
     pub fn get_pointer(&self) -> Option<ShelleyAddressPointer> {

common/src/lib.rs

@@ -15,6 +15,7 @@ pub mod protocol_params;
 pub mod queries;
 pub mod rational_number;
 pub mod resolver;
+pub mod rest_error;
 pub mod rest_helper;
 pub mod serialization;
 pub mod snapshot;

common/src/messages.rs

@@ -311,6 +311,7 @@ pub enum CardanoMessage {
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub enum SnapshotMessage {
+    Startup, // subscribers should listen for incremental snapshot data
     Bootstrap(SnapshotStateMessage),
     DumpRequest(SnapshotDumpMessage),
     Dump(SnapshotStateMessage),

common/src/queries/accounts.rs

@@ -1,5 +1,6 @@
 use std::collections::HashMap;
 
+use crate::queries::errors::QueryError;
 use crate::{
     DRepChoice, PoolId, PoolLiveStakeInfo, RewardType, ShelleyAddress, StakeAddress, TxIdentifier,
 };
@@ -78,9 +79,7 @@ pub enum AccountsStateQueryResponse {
     // DReps-related responses
     DrepDelegators(DrepDelegators),
     AccountsDrepDelegationsMap(HashMap<StakeAddress, Option<DRepChoice>>),
-
-    NotFound,
-    Error(String),
+    Error(QueryError),
 }
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]

common/src/queries/addresses.rs

@@ -1,3 +1,4 @@
+use crate::queries::errors::QueryError;
 use crate::{
     Address, AddressTotals, NativeAssets, ShelleyAddress, TxIdentifier, UTxOIdentifier, ValueDelta,
 };
@@ -27,6 +28,5 @@ pub enum AddressStateQueryResponse {
     AddressesAssets(NativeAssets),
     AddressesTotals(ValueDelta),
     AddressesUTxOs(Vec<UTxOIdentifier>),
-    NotFound,
-    Error(String),
+    Error(QueryError),
 }

common/src/queries/assets.rs

@@ -1,3 +1,4 @@
+use crate::queries::errors::QueryError;
 use crate::{
     AssetAddressEntry, AssetInfoRecord, AssetMintRecord, AssetName, PolicyAsset, PolicyId,
     TxIdentifier,
@@ -36,6 +37,5 @@ pub enum AssetsStateQueryResponse {
     AssetAddresses(AssetAddresses),
     AssetTransactions(AssetTransactions),
     PolicyIdAssets(PolicyAssets),
-    NotFound,
-    Error(String),
+    Error(QueryError),
 }

common/src/queries/blocks.rs

@@ -1,3 +1,4 @@
+use crate::queries::errors::QueryError;
 use crate::{
     queries::misc::Order,
     serialization::{Bech32Conversion, Bech32WithHrp},
@@ -95,8 +96,7 @@ pub enum BlocksStateQueryResponse {
     BlockHashes(BlockHashes),
     TransactionHashes(TransactionHashes),
     UTxOHashes(UTxOHashes),
-    NotFound,
-    Error(String),
+    Error(QueryError),
 }
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]

common/src/queries/epochs.rs

@@ -1,3 +1,4 @@
+use crate::queries::errors::QueryError;
 use crate::{messages::EpochActivityMessage, protocol_params::ProtocolParams, PoolId};
 
 pub const DEFAULT_EPOCHS_QUERY_TOPIC: (&str, &str) =
@@ -23,9 +24,7 @@ pub enum EpochsStateQueryResponse {
     EpochStakeDistribution(EpochStakeDistribution),
     EpochStakeDistributionByPool(EpochStakeDistributionByPool),
     LatestEpochBlocksMintedByPool(u64),
-
-    NotFound,
-    Error(String),
+    Error(QueryError),
 }
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]

common/src/queries/errors.rs

@@ -0,0 +1,58 @@
+use serde::{Deserialize, Serialize};
+use thiserror::Error;
+
+/// Common error type for all state query responses
+#[derive(Debug, Clone, Error, Serialize, Deserialize)]
+pub enum QueryError {
+    /// The requested resource was not found
+    #[error("Not found: {resource}")]
+    NotFound { resource: String },
+
+    /// An error occurred while processing the query
+    #[error("Internal error: {message}")]
+    Internal { message: String },
+
+    /// Storage backend is disabled in configuration
+    #[error("{storage_type} storage is not enabled")]
+    StorageDisabled { storage_type: String },
+
+    /// Invalid request parameters
+    #[error("Invalid request: {message}")]
+    InvalidRequest { message: String },
+
+    /// Query variant is not implemented yet
+    #[error("Query not implemented: {query}")]
+    NotImplemented { query: String },
+}
+
+impl QueryError {
+    pub fn not_found(resource: impl Into<String>) -> Self {
+        Self::NotFound {
+            resource: resource.into(),
+        }
+    }
+
+    pub fn internal_error(message: impl Into<String>) -> Self {
+        Self::Internal {
+            message: message.into(),
+        }
+    }
+
+    pub fn storage_disabled(storage_type: impl Into<String>) -> Self {
+        Self::StorageDisabled {
+            storage_type: storage_type.into(),
+        }
+    }
+
+    pub fn invalid_request(message: impl Into<String>) -> Self {
+        Self::InvalidRequest {
+            message: message.into(),
+        }
+    }
+
+    pub fn not_implemented(query: impl Into<String>) -> Self {
+        Self::NotImplemented {
+            query: query.into(),
+        }
+    }
+}