feat: get_queue_elements for input+output queues

cli/src/commands/test-validator/index.ts

@@ -69,18 +69,18 @@ class SetupCommand extends Command {
       default: 8784,
       exclusive: ["skip-indexer"],
     }),
-    "prover-port": Flags.integer({
-      description: "Enable Light Prover server on this port.",
-      required: false,
-      default: 3001,
-      exclusive: ["skip-prover"],
-    }),
     "grpc-port": Flags.integer({
       description: "Enable Photon indexer gRPC on this port.",
       required: false,
       default: 50051,
       exclusive: ["skip-indexer"],
     }),
+    "prover-port": Flags.integer({
+      description: "Enable Light Prover server on this port.",
+      required: false,
+      default: 3001,
+      exclusive: ["skip-prover"],
+    }),
     "limit-ledger-size": Flags.integer({
       description: "Keep this amount of shreds in root slots.",
       required: false,

cli/src/utils/initTestEnv.ts

@@ -1,4 +1,3 @@
-import { airdropSol } from "@lightprotocol/stateless.js";
 import { getConfig, getPayer, setAnchorProvider, setConfig } from "./utils";
 import {
   BASE_PATH,
@@ -131,9 +130,9 @@ export async function initTestEnv({
     await startIndexer(
       `http://127.0.0.1:${rpcPort}`,
       indexerPort,
+      grpcPort,
       checkPhotonVersion,
       photonDatabaseUrl,
-      grpcPort,
     );
   }
 

cli/src/utils/process.ts

@@ -212,6 +212,10 @@ export function spawnBinary(command: string, args: string[] = []) {
       stdio: ["ignore", out, err],
       shell: false,
       detached: true,
+      env: {
+        ...process.env,
+        RUST_LOG: process.env.RUST_LOG || "debug",
+      },
     });
 
     spawnedProcess.on("close", async (code) => {

cli/src/utils/processPhotonIndexer.ts

@@ -39,9 +39,9 @@ function getPhotonInstallMessage(): string {
 export async function startIndexer(
   rpcUrl: string,
   indexerPort: number,
+  grpcPort: number = 50051,
   checkPhotonVersion: boolean = true,
   photonDatabaseUrl?: string,
-  grpcPort: number = 50051,
 ) {
   await killIndexer();
   const resolvedOrNull = which.sync("photon", { nothrow: true });

forester-utils/src/instructions/state_batch_append.rs

@@ -9,7 +9,6 @@ use light_batched_merkle_tree::{
 use light_client::{indexer::Indexer, rpc::Rpc};
 use light_compressed_account::instruction_data::compressed_proof::CompressedProof;
 use light_hasher::bigint::bigint_to_be_bytes_array;
-use light_merkle_tree_metadata::QueueType;
 use light_prover_client::{
     proof_client::ProofClient,
     proof_types::batch_append::{get_batch_append_inputs, BatchAppendsCircuitInputs},
@@ -120,18 +119,19 @@ pub async fn get_append_instruction_stream<'a, R: Rpc>(
                 indexer
                     .get_queue_elements(
                         merkle_tree_pubkey.to_bytes(),
-                        QueueType::OutputStateV2,
-                        zkp_batch_size,
                         next_queue_index,
+                        Some(zkp_batch_size),
+                        None,
+                        None,
                         None,
                     )
                     .await
             };
 
             let (batch_elements, batch_first_queue_idx) = match queue_elements_result {
                 Ok(res) => {
-                    let items = res.value.elements;
-                    let first_idx = res.value.first_value_queue_index;
+                    let items = res.value.output_queue_elements.unwrap_or_default();
+                    let first_idx = res.value.output_queue_index;
                     if items.len() != zkp_batch_size as usize {
                         warn!(
                             "Got {} elements but expected {}, stopping",
@@ -238,6 +238,5 @@ pub async fn get_append_instruction_stream<'a, R: Rpc>(
             yield Ok(proofs_buffer);
         }
     };
-
     Ok((Box::pin(stream), zkp_batch_size))
 }

forester-utils/src/instructions/state_batch_nullify.rs

@@ -9,7 +9,6 @@ use light_batched_merkle_tree::{
 use light_client::{indexer::Indexer, rpc::Rpc};
 use light_compressed_account::instruction_data::compressed_proof::CompressedProof;
 use light_hasher::bigint::bigint_to_be_bytes_array;
-use light_merkle_tree_metadata::QueueType;
 use light_prover_client::{
     proof_client::ProofClient,
     proof_types::batch_update::{get_batch_update_inputs, BatchUpdateCircuitInputs},
@@ -113,18 +112,19 @@ pub async fn get_nullify_instruction_stream<'a, R: Rpc>(
                 let indexer = connection.indexer_mut()?;
                 indexer.get_queue_elements(
                     merkle_tree_pubkey.to_bytes(),
-                    QueueType::InputStateV2,
-                    zkp_batch_size,
+                    None,
+                    None,
                     next_queue_index,
+                    Some(zkp_batch_size),
                     None,
                 )
                 .await
             };
 
             let (batch_elements, batch_first_queue_idx) = match queue_elements_result {
                 Ok(res) => {
-                    let items = res.value.elements;
-                    let first_idx = res.value.first_value_queue_index;
+                    let items = res.value.input_queue_elements.unwrap_or_default();
+                    let first_idx = res.value.input_queue_index;
                     if items.len() != zkp_batch_size as usize {
                         warn!(
                             "Got {} elements but expected {}, stopping",

forester/src/config.rs

@@ -83,6 +83,8 @@ pub struct GeneralConfig {
     pub skip_v2_state_trees: bool,
     pub skip_v2_address_trees: bool,
     pub tree_id: Option<Pubkey>,
+    pub sleep_after_processing_ms: u64,
+    pub sleep_when_idle_ms: u64,
 }
 
 impl Default for GeneralConfig {
@@ -96,6 +98,8 @@ impl Default for GeneralConfig {
             skip_v2_state_trees: false,
             skip_v2_address_trees: false,
             tree_id: None,
+            sleep_after_processing_ms: 10_000,
+            sleep_when_idle_ms: 45_000,
         }
     }
 }
@@ -111,6 +115,8 @@ impl GeneralConfig {
             skip_v2_state_trees: true,
             skip_v2_address_trees: false,
             tree_id: None,
+            sleep_after_processing_ms: 50,
+            sleep_when_idle_ms: 100,
         }
     }
 
@@ -124,6 +130,8 @@ impl GeneralConfig {
             skip_v2_state_trees: false,
             skip_v2_address_trees: true,
             tree_id: None,
+            sleep_after_processing_ms: 50,
+            sleep_when_idle_ms: 100,
         }
     }
 }
@@ -276,6 +284,8 @@ impl ForesterConfig {
                     .tree_id
                     .as_ref()
                     .and_then(|id| Pubkey::from_str(id).ok()),
+                sleep_after_processing_ms: 10_000,
+                sleep_when_idle_ms: 45_000,
             },
             rpc_pool_config: RpcPoolConfig {
                 max_size: args.rpc_pool_size,
@@ -332,6 +342,8 @@ impl ForesterConfig {
                 skip_v1_address_trees: false,
                 skip_v2_address_trees: false,
                 tree_id: None,
+                sleep_after_processing_ms: 10_000,
+                sleep_when_idle_ms: 45_000,
             },
             rpc_pool_config: RpcPoolConfig {
                 max_size: 10,

forester/src/epoch_manager.rs

@@ -1260,11 +1260,6 @@ impl<R: Rpc> EpochManager<R> {
             forester_slot_details.start_solana_slot,
         )
         .await?;
-        info!(
-            "Slot {} started, beginning processing",
-            forester_slot_details.slot
-        );
-
         let mut estimated_slot = self.slot_tracker.estimated_current_slot();
 
         'inner_processing_loop: loop {
@@ -1334,9 +1329,9 @@ impl<R: Rpc> EpochManager<R> {
             estimated_slot = self.slot_tracker.estimated_current_slot();
 
             let sleep_duration_ms = if items_processed_this_iteration > 0 {
-                10_000
+                self.config.general_config.sleep_after_processing_ms
             } else {
-                45_000
+                self.config.general_config.sleep_when_idle_ms
             };
 
             tokio::time::sleep(Duration::from_millis(sleep_duration_ms)).await;
@@ -2184,6 +2179,8 @@ mod tests {
                 skip_v2_state_trees: skip_v2_state,
                 skip_v2_address_trees: skip_v2_address,
                 tree_id: None,
+                sleep_after_processing_ms: 50,
+                sleep_when_idle_ms: 100,
             },
             rpc_pool_config: Default::default(),
             registry_pubkey: Pubkey::default(),

forester/tests/e2e_test.rs

@@ -231,6 +231,8 @@ async fn e2e_test() {
             skip_v1_address_trees: false,
             skip_v2_address_trees: false,
             tree_id: None,
+            sleep_after_processing_ms: 50,
+            sleep_when_idle_ms: 100,
         },
         rpc_pool_config: RpcPoolConfig {
             max_size: 50,

forester/tests/legacy/test_utils.rs

@@ -94,6 +94,9 @@ pub fn forester_config() -> ForesterConfig {
             skip_v2_state_trees: false,
             skip_v1_address_trees: false,
             skip_v2_address_trees: false,
+            tree_id: None,
+            sleep_after_processing_ms: 50,
+            sleep_when_idle_ms: 100,
         },
         rpc_pool_config: RpcPoolConfig {
             max_size: 50,

forester/tests/test_utils.rs

@@ -109,6 +109,8 @@ pub fn forester_config() -> ForesterConfig {
             skip_v1_address_trees: false,
             skip_v2_address_trees: false,
             tree_id: None,
+            sleep_after_processing_ms: 50,
+            sleep_when_idle_ms: 100,
         },
         rpc_pool_config: RpcPoolConfig {
             max_size: 50,

program-tests/utils/src/e2e_test_env.rs

@@ -121,7 +121,6 @@ use light_hasher::{bigint::bigint_to_be_bytes_array, Poseidon};
 use light_indexed_merkle_tree::{
     array::IndexedArray, reference::IndexedMerkleTree, HIGHEST_ADDRESS_PLUS_ONE,
 };
-use light_merkle_tree_metadata::QueueType;
 use light_program_test::{
     accounts::{
         state_tree::create_state_merkle_tree_and_queue_account, test_accounts::TestAccounts,
@@ -746,15 +745,16 @@ where
                                     .indexer
                                     .get_queue_elements(
                                         merkle_tree_pubkey.to_bytes(),
-                                        QueueType::AddressV2,
-                                        batch.batch_size as u16,
+                                        None,
+                                        Some(batch.batch_size as u16),
+                                        None,
                                         None,
                                         None,
                                     )
                                     .await
                                     .unwrap();
                                 let addresses =
-                                    addresses.value.elements.iter().map(|x| x.account_hash).collect::<Vec<_>>();
+                                    addresses.value.output_queue_elements.unwrap_or_default().iter().map(|x| x.account_hash).collect::<Vec<_>>();
                                 // // local_leaves_hash_chain is only used for a test assertion.
                                 // let local_nullifier_hash_chain = create_hash_chain_from_array(&addresses);
                                 // assert_eq!(leaves_hash_chain, local_nullifier_hash_chain);

program-tests/utils/src/test_batch_forester.rs

@@ -23,7 +23,6 @@ use light_batched_merkle_tree::{
 use light_client::rpc::{Rpc, RpcError};
 use light_compressed_account::{
     hash_chain::create_hash_chain_from_slice, instruction_data::compressed_proof::CompressedProof,
-    QueueType,
 };
 use light_hasher::{bigint::bigint_to_be_bytes_array, Poseidon};
 use light_prover_client::{
@@ -654,16 +653,18 @@ pub async fn create_batch_update_address_tree_instruction_data_with_proof<R: Rpc
     let addresses = indexer
         .get_queue_elements(
             merkle_tree_pubkey.to_bytes(),
-            QueueType::AddressV2,
-            batch.zkp_batch_size as u16,
+            None,
+            Some(batch.zkp_batch_size as u16),
+            None,
             None,
             None,
         )
         .await
         .unwrap();
     let addresses = addresses
         .value
-        .elements
+        .output_queue_elements
+        .unwrap_or_default()
         .iter()
         .map(|x| x.account_hash)
         .collect::<Vec<[u8; 32]>>();

prover/server/main.go

@@ -21,7 +21,7 @@ import (
 	"github.com/urfave/cli/v2"
 )
 
-const Version = "2.0.0"
+const Version = "2.0.6"
 
 func main() {
 	runCli()

scripts/devenv/versions.sh

@@ -13,7 +13,7 @@ export SOLANA_VERSION="2.2.15"
 export ANCHOR_VERSION="0.31.1"
 export JQ_VERSION="1.8.0"
 export PHOTON_VERSION="0.51.0"
-export PHOTON_COMMIT="06862b290f32025bc150f82a4acba4961ee24178"
+export PHOTON_COMMIT="94b3688b08477668bb946a689b0267319f5c1ae1"
 export REDIS_VERSION="8.0.1"
 
 export ANCHOR_TAG="anchor-v${ANCHOR_VERSION}"

sdk-libs/client/src/indexer/indexer_trait.rs

@@ -1,5 +1,4 @@
 use async_trait::async_trait;
-use light_merkle_tree_metadata::QueueType;
 use solana_pubkey::Pubkey;
 
 use super::{
@@ -187,16 +186,18 @@ pub trait Indexer: std::marker::Send + std::marker::Sync {
     // TODO: in different pr:
     //      replace num_elements & start_queue_index with PaginatedOptions
     //      - return type should be ItemsWithCursor
-    /// Returns queue elements from the queue with the given merkle tree pubkey. For input
-    /// queues account compression program does not store queue elements in the
+    /// Returns queue elements from the queue with the given merkle tree pubkey.
+    /// Can fetch from output queue (append), input queue (nullify), or both atomically.
+    /// For input queues account compression program does not store queue elements in the
     /// account data but only emits these in the public transaction event. The
     /// indexer needs the queue elements to create batch update proofs.
     async fn get_queue_elements(
         &mut self,
         merkle_tree_pubkey: [u8; 32],
-        queue_type: QueueType,
-        num_elements: u16,
-        start_queue_index: Option<u64>,
+        output_queue_start_index: Option<u64>,
+        output_queue_limit: Option<u16>,
+        input_queue_start_index: Option<u64>,
+        input_queue_limit: Option<u16>,
         config: Option<IndexerRpcConfig>,
     ) -> Result<Response<QueueElementsResult>, IndexerError>;