[accumulator-updater 5/x] Single PDA (#760)

* feat(accumulator-updater): write accumulator messages into one account

* chore(accumulator-updater): clean up commented out code

* chore(accumulator-updater): remove unused feature for cargo

* chore(accumulator-updater): minor comment fix

* feat(accumulator-updater): update implementation of AccumulatorInput header

Removed InputIndex in header, updated to use end_offsets & include header_len

* feat(accumulator-updater): add MessageHeader to all PriceMessages and include in serialization

* fix(accumulator-updater): fix AccumulatorInput size & unit tests

* feat(accumulator-updater): update put_all to write as much data as possible

put_all will write up to the limit of the accumulator_input.data.len(). this ix must not fail even
when the data being written exceeds the size of the account. in this situation, we will write as
many *COMPLETE* messages as possible

* docs(accumulator-updater): update put_all ix documentation

* chore(accumulator-updater): addressed PR feedback

fixed comments, added test, renamed accumulatorInput.data to messages

* fix(accumulator-updater): fix ts test

Author: swimricky

accumulator_updater/programs/accumulator_updater/Cargo.toml

@@ -16,6 +16,6 @@ cpi = ["no-entrypoint"]
 default = []
 
 [dependencies]
-anchor-lang = "0.27.0"
+anchor-lang = { version = "0.27.0" }
 # needed for the new #[account(zero_copy)] in anchor 0.27.0
 bytemuck = { version = "1.4.0", features = ["derive", "min_const_generics"]}

accumulator_updater/programs/accumulator_updater/src/instructions/put_all.rs

@@ -12,99 +12,73 @@ use {
     },
 };
 
-#[derive(AnchorSerialize, AnchorDeserialize)]
-pub struct InputSchemaAndData {
-    pub schema: u8,
-    pub data:   Vec<u8>,
-}
 
+pub const ACCUMULATOR: &[u8; 11] = b"accumulator";
 
 pub fn put_all<'info>(
     ctx: Context<'_, '_, '_, 'info, PutAll<'info>>,
     base_account_key: Pubkey,
-    values: Vec<InputSchemaAndData>,
+    messages: Vec<Vec<u8>>,
 ) -> Result<()> {
     let cpi_caller = ctx.accounts.whitelist_verifier.is_allowed()?;
-    let account_infos = ctx.remaining_accounts;
-    require_eq!(account_infos.len(), values.len());
+    let accumulator_input_ai = ctx
+        .remaining_accounts
+        .first()
+        .ok_or(AccumulatorUpdaterError::AccumulatorInputNotProvided)?;
 
-    let rent = Rent::get()?;
-    let (mut initialized, mut updated) = (vec![], vec![]);
+    let loader;
 
-    for (
-        ai,
-        InputSchemaAndData {
-            schema: account_schema,
-            data: account_data,
-        },
-    ) in account_infos.iter().zip(values)
     {
-        let bump = if is_uninitialized_account(ai) {
-            let seeds = &[
+        let accumulator_input = &mut (if is_uninitialized_account(accumulator_input_ai) {
+            let (pda, bump) = Pubkey::find_program_address(
+                &[
+                    cpi_caller.as_ref(),
+                    ACCUMULATOR.as_ref(),
+                    base_account_key.as_ref(),
+                ],
+                &crate::ID,
+            );
+            require_keys_eq!(accumulator_input_ai.key(), pda);
+            let signer_seeds = &[
                 cpi_caller.as_ref(),
-                b"accumulator".as_ref(),
+                ACCUMULATOR.as_ref(),
                 base_account_key.as_ref(),
-                &account_schema.to_le_bytes(),
+                &[bump],
             ];
-            let (pda, bump) = Pubkey::find_program_address(seeds, &crate::ID);
-            require_keys_eq!(ai.key(), pda);
-
-
-            //TODO: Update this with serialization logic
-            // 8 for anchor discriminator
-            let accumulator_size = 8 + AccumulatorInput::size(&account_data);
             PutAll::create_account(
-                ai,
-                accumulator_size,
+                accumulator_input_ai,
+                8 + AccumulatorInput::INIT_SPACE,
                 &ctx.accounts.payer,
-                &[
-                    cpi_caller.as_ref(),
-                    b"accumulator".as_ref(),
-                    base_account_key.as_ref(),
-                    &account_schema.to_le_bytes(),
-                    &[bump],
-                ],
-                &rent,
+                signer_seeds,
                 &ctx.accounts.system_program,
             )?;
-            initialized.push(ai.key());
-
-            bump
-        } else {
-            let accumulator_input = <AccumulatorInput as AccountDeserialize>::try_deserialize(
-                &mut &**ai.try_borrow_mut_data()?,
+            loader = AccountLoader::<AccumulatorInput>::try_from_unchecked(
+                &crate::ID,
+                accumulator_input_ai,
             )?;
-            {
-                // TODO: allow re-sizing?
-                require_gte!(
-                    accumulator_input.data.len(),
-                    account_data.len(),
-                    AccumulatorUpdaterError::CurrentDataLengthExceeded
-                );
-
-                accumulator_input.validate(
-                    ai.key(),
-                    cpi_caller,
-                    base_account_key,
-                    account_schema,
-                )?;
-            }
-
+            let mut accumulator_input = loader.load_init()?;
+            accumulator_input.header = AccumulatorHeader::new(bump);
+            accumulator_input
+        } else {
+            loader = AccountLoader::<AccumulatorInput>::try_from(accumulator_input_ai)?;
+            let mut accumulator_input = loader.load_mut()?;
+            accumulator_input.header.set_version();
+            accumulator_input
+        });
+        // note: redundant for uninitialized code path but safer to check here.
+        // compute budget cost should be minimal
+        accumulator_input.validate(accumulator_input_ai.key(), cpi_caller, base_account_key)?;
+
+
+        let (num_msgs, num_bytes) = accumulator_input.put_all(&messages);
+        if num_msgs != messages.len() {
+            msg!("unable to fit all messages in accumulator input account. Wrote {}/{} messages and {} bytes", num_msgs, messages.len(), num_bytes);
+        }
+    }
 
-            updated.push(ai.key());
-            accumulator_input.header.bump
-        };
 
-        let accumulator_input =
-            AccumulatorInput::new(AccumulatorHeader::new(bump, account_schema), account_data);
-        accumulator_input.persist(ai)?;
-    }
+    loader.exit(&crate::ID)?;
 
-    msg!(
-        "[emit-updates]: initialized: {:?}, updated: {:?}",
-        initialized,
-        updated
-    );
     Ok(())
 }
 
@@ -113,12 +87,13 @@ pub fn is_uninitialized_account(ai: &AccountInfo) -> bool {
 }
 
 #[derive(Accounts)]
+#[instruction( base_account_key: Pubkey)]
 pub struct PutAll<'info> {
     #[account(mut)]
     pub payer:              Signer<'info>,
     pub whitelist_verifier: WhitelistVerifier<'info>,
     pub system_program:     Program<'info, System>,
-    // remaining_accounts:  - [AccumulatorInput PDAs]
+    // remaining_accounts:  - [AccumulatorInput PDA]
 }
 
 impl<'info> PutAll<'info> {
@@ -127,10 +102,9 @@ impl<'info> PutAll<'info> {
         space: usize,
         payer: &AccountInfo<'a>,
         seeds: &[&[u8]],
-        rent: &Rent,
         system_program: &AccountInfo<'a>,
     ) -> Result<()> {
-        let lamports = rent.minimum_balance(space);
+        let lamports = Rent::get()?.minimum_balance(space);
 
         system_program::create_account(
             CpiContext::new_with_signer(

accumulator_updater/programs/accumulator_updater/src/lib.rs

@@ -52,24 +52,36 @@ pub mod accumulator_updater {
     }
 
 
-    /// Create or update inputs for the Accumulator. Each input is written
-    /// into a separate PDA account derived with
-    /// seeds = [cpi_caller, b"accumulator", base_account_key, schema]
+    /// Insert messages/inputs for the Accumulator. All inputs derived from the
+    /// `base_account_key` will go into the same PDA. The PDA is derived with
+    /// seeds = [cpi_caller, b"accumulator", base_account_key]
     ///
-    /// The caller of this instruction must pass those PDAs
-    /// while calling this function in the same order as elements.
     ///
     ///
     /// * `base_account_key`    - Pubkey of the original account the
-    ///                         AccumulatorInput(s) are derived from
-    /// * `values`              - Vec of (schema, account_data) in same respective
-    ///                           order `ctx.remaining_accounts`
+    ///                           AccumulatorInput is derived from
+    /// * `messages`            - Vec of vec of bytes, each representing a message
+    ///                           to be hashed and accumulated
+    ///
+    /// This ix will write as many of the messages up to the length
+    /// of the `accumulator_input.data`.
+    /// If `accumulator_input.data.len() < messages.map(|x| x.len()).sum()`
+    /// then the remaining messages will be ignored.
+    ///
+    /// The current implementation assumes that each invocation of this
+    /// ix is independent of any previous invocations. It will overwrite
+    /// any existing contents.
+    ///
+    /// TODO:
+    ///     - try handling re-allocation of the accumulator_input space
+    ///     - handle updates ("paging/batches of messages")
+    ///
     pub fn put_all<'info>(
         ctx: Context<'_, '_, '_, 'info, PutAll<'info>>,
         base_account_key: Pubkey,
-        values: Vec<InputSchemaAndData>,
+        messages: Vec<Vec<u8>>,
     ) -> Result<()> {
-        instructions::put_all(ctx, base_account_key, values)
+        instructions::put_all(ctx, base_account_key, messages)
     }
 }
 
@@ -128,4 +140,6 @@ pub enum AccumulatorUpdaterError {
     CurrentDataLengthExceeded,
     #[msg("Accumulator Input not writable")]
     AccumulatorInputNotWritable,
+    #[msg("Accumulator Input not provided")]
+    AccumulatorInputNotProvided,
 }

accumulator_updater/programs/accumulator_updater/src/macros.rs

@@ -5,7 +5,6 @@ macro_rules! accumulator_input_seeds {
             $cpi_caller_pid.as_ref(),
             b"accumulator".as_ref(),
             $base_account.as_ref(),
-            &$accumulator_input.header.account_schema.to_le_bytes(),
             &[$accumulator_input.header.bump],
         ]
     };