Merge branch 'master' into thlorenz/subscription-metrics

* master:
  hotfix: unregistered metric COMMITTOR_INTENT_CU_USAGE (#612)
  feat: add metrics for intent execution time (#607)
  fix: don't increment subscription id for dups (#610)
  chore: broaden dependency range on the magic program api (#609)
  feat: parallel CI integration tests (#605)
  Handle uncleaned buffers + generalized tx send retries (#597)
  fix: support binary opcode (#604)
  fix: correct log imports (#606)

Author: thlorenz

.github/workflows/ci-test-integration.yml

@@ -11,12 +11,13 @@ on:
     types: [opened, reopened, synchronize, ready_for_review]
 
 jobs:
-  run_make_ci_test:
-    if: github.event.pull_request.draft == false
+  build:
+    if: github.event_name != 'pull_request' || !github.event.pull_request.draft
     runs-on: extra-large
+    name: Build Project
     steps:
       - name: Checkout this magicblock-validator
-        uses: actions/checkout@v2
+        uses: actions/checkout@v5
         with:
           path: magicblock-validator
 
@@ -29,11 +30,54 @@ jobs:
 
       - uses: ./magicblock-validator/.github/actions/setup-solana
 
-      - name: Run integration tests
+      - name: Build project and test programs
+        run: |
+          cargo build --locked
+          make -C test-integration programs
+        shell: bash
+        working-directory: magicblock-validator
+
+  run_integration_tests:
+    needs: build
+    runs-on: extra-large
+    strategy:
+      matrix:
+        batch_tests:
+          - "schedulecommit"
+          - "chainlink"
+          - "cloning"
+          - "restore_ledger"
+          - "magicblock_api"
+          - "config"
+          - "table_mania"
+          - "committor"
+          - "pubsub"
+          - "schedule_intents"
+          - "task-scheduler"
+      fail-fast: false
+    name: Integration Tests - ${{ matrix.batch_tests }}
+    steps:
+      - name: Checkout this magicblock-validator
+        uses: actions/checkout@v5
+        with:
+          path: magicblock-validator
+
+      - uses: ./magicblock-validator/.github/actions/setup-build-env
+        with:
+          build_cache_key_name: "magicblock-validator-ci-test-integration-v000"
+          rust_toolchain_release: "1.84.1"
+          github_access_token: ${{ secrets.GH_PERSONAL_ACCESS_TOKEN }}
+          github_token: ${{ secrets.GITHUB_TOKEN }}
+
+      - uses: ./magicblock-validator/.github/actions/setup-solana
+
+      - name: Run integration tests - ${{ matrix.batch_tests }}
         run: |
           sudo prlimit --pid $$ --nofile=1048576:1048576
           sudo sysctl fs.inotify.max_user_instances=1280
           sudo sysctl fs.inotify.max_user_watches=655360
           make ci-test-integration
         shell: bash
         working-directory: magicblock-validator
+        env:
+          RUN_TESTS: ${{ matrix.batch_tests }}

.github/workflows/publish-packages.yml

@@ -1,4 +1,4 @@
-name: Publish ephemeral validator packages
+name: Publish ephemeral validator packages and crates
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.ref }}
@@ -213,4 +213,20 @@ jobs:
             npm publish --access public
           fi
         env:
-          NODE_AUTH_TOKEN: ${{ secrets.NPM_TOKEN }}
+          NODE_AUTH_TOKEN: ${{ secrets.NPM_TOKEN }}
+
+      - name: cargo publish
+        working-directory: magicblock-magic-program-api/
+        run: |
+          DRY_RUN_FLAG=""
+          if [ "${DRY_RUN}" = "true" ]; then
+            DRY_RUN_FLAG="--dry-run"
+          fi
+
+          if [ "${DRY_RUN}" = "true" ]; then
+            NO_VERIFY_FLAG="--no-verify"
+          fi      
+          cargo publish $DRY_RUN_FLAG --manifest-path=./Cargo.toml --token $CRATES_TOKEN $NO_VERIFY_FLAG
+        env:
+          CRATES_TOKEN: ${{ secrets.CRATES_TOKEN }}
+          DRY_RUN: ${{ env.DRY_RUN }}

Makefile

@@ -52,8 +52,7 @@ ci-test-unit:
 	RUST_BACKTRACE=1 cargo $(CARGO_TEST_NOCAP)
 
 ci-test-integration:
-	cargo build --locked && \
-	$(MAKE) -C $(DIR)/test-integration test
+	RUN_TESTS=$(RUN_TESTS) $(MAKE) -C $(DIR)/test-integration test
 
 ## NOTE: We're getting the following error in github CI when trying to use
 #  nightly Rust. Until that is fixed we have to use stable to verify format.

magicblock-aperture/src/encoder.rs

@@ -1,3 +1,5 @@
+use std::fmt::Debug;
+
 use hyper::body::Bytes;
 use json::Serialize;
 use magicblock_core::{
@@ -20,7 +22,7 @@ use crate::{
 
 /// An abstraction trait over types which specialize in turning various
 /// websocket notification payload types into sequence of bytes
-pub(crate) trait Encoder: Ord + Eq + Clone {
+pub(crate) trait Encoder: Ord + Eq + Clone + Debug {
     type Data;
     fn encode(
         &self,
@@ -64,7 +66,7 @@ impl From<UiAccountEncoding> for AccountEncoder {
 }
 
 /// A `programSubscribe` payload encoder
-#[derive(PartialEq, PartialOrd, Ord, Eq, Clone)]
+#[derive(PartialEq, PartialOrd, Ord, Eq, Clone, Debug)]
 pub struct ProgramAccountEncoder {
     pub encoder: AccountEncoder,
     pub filters: ProgramFilters,
@@ -106,7 +108,7 @@ impl Encoder for ProgramAccountEncoder {
 }
 
 /// A `signatureSubscribe` payload encoder
-#[derive(PartialEq, PartialOrd, Ord, Eq, Clone)]
+#[derive(PartialEq, PartialOrd, Ord, Eq, Clone, Debug)]
 pub(crate) struct TransactionResultEncoder;
 
 impl Encoder for TransactionResultEncoder {
@@ -130,7 +132,7 @@ impl Encoder for TransactionResultEncoder {
 }
 
 /// A `logsSubscribe` payload encoder
-#[derive(PartialEq, PartialOrd, Ord, Eq, Clone)]
+#[derive(PartialEq, PartialOrd, Ord, Eq, Clone, Debug)]
 pub(crate) enum TransactionLogsEncoder {
     All,
     Mentions(Pubkey),
@@ -172,7 +174,7 @@ impl Encoder for TransactionLogsEncoder {
 }
 
 /// A `slotSubscribe` payload encoder
-#[derive(PartialEq, PartialOrd, Ord, Eq, Clone)]
+#[derive(PartialEq, PartialOrd, Ord, Eq, Clone, Debug)]
 pub(crate) struct SlotEncoder;
 
 impl Encoder for SlotEncoder {

magicblock-aperture/src/requests/websocket/account_subscribe.rs

@@ -32,8 +32,10 @@ impl WsDispatcher {
             .subscribe_to_account(pubkey, encoder, self.chan.clone())
             .await;
 
+        let result = SubResult::SubId(handle.id);
         // Store the cleanup handle to manage the subscription's lifecycle.
-        self.unsubs.insert(handle.id, handle.cleanup);
-        Ok(SubResult::SubId(handle.id))
+        self.register_unsub(handle);
+
+        Ok(result)
     }
 }

magicblock-aperture/src/requests/websocket/log_subscribe.rs

@@ -35,7 +35,10 @@ impl WsDispatcher {
             .subscriptions
             .subscribe_to_logs(encoder, self.chan.clone());
 
-        self.unsubs.insert(handle.id, handle.cleanup);
-        Ok(SubResult::SubId(handle.id))
+        let result = SubResult::SubId(handle.id);
+        // Store the cleanup handle to manage the subscription's lifecycle.
+        self.register_unsub(handle);
+
+        Ok(result)
     }
 }

magicblock-aperture/src/requests/websocket/program_subscribe.rs

@@ -43,7 +43,10 @@ impl WsDispatcher {
             .subscribe_to_program(pubkey, encoder, self.chan.clone())
             .await;
 
-        self.unsubs.insert(handle.id, handle.cleanup);
-        Ok(SubResult::SubId(handle.id))
+        let result = SubResult::SubId(handle.id);
+        // Store the cleanup handle to manage the subscription's lifecycle.
+        self.register_unsub(handle);
+
+        Ok(result)
     }
 }

magicblock-aperture/src/requests/websocket/slot_subscribe.rs

@@ -7,7 +7,10 @@ impl WsDispatcher {
     /// each time the validator advances to a new slot.
     pub(crate) fn slot_subscribe(&mut self) -> RpcResult<SubResult> {
         let handle = self.subscriptions.subscribe_to_slot(self.chan.clone());
-        self.unsubs.insert(handle.id, handle.cleanup);
-        Ok(SubResult::SubId(handle.id))
+        let result = SubResult::SubId(handle.id);
+        // Store the cleanup handle to manage the subscription's lifecycle.
+        self.register_unsub(handle);
+
+        Ok(result)
     }
 }

magicblock-aperture/src/server/websocket/connection.rs

@@ -112,7 +112,7 @@ impl ConnectionHandler {
                     // Reschedule the next ping
                     next_ping.as_mut().reset(Instant::now() + PING_PERIOD);
 
-                    if frame.opcode != OpCode::Text {
+                    if frame.opcode != OpCode::Text && frame.opcode != OpCode::Binary {
                         continue;
                     }
 

magicblock-aperture/src/server/websocket/dispatch.rs

@@ -12,7 +12,9 @@ use crate::{
     requests::{JsonRpcWsMethod, JsonWsRequest},
     state::{
         signatures::SignaturesExpirer,
-        subscriptions::{CleanUp, SubscriptionID, SubscriptionsDb},
+        subscriptions::{
+            CleanUp, SubscriptionHandle, SubscriptionID, SubscriptionsDb,
+        },
         transactions::TransactionsCache,
     },
     RpcResult,
@@ -123,6 +125,16 @@ impl WsDispatcher {
         let success = self.unsubs.remove(&id).is_some();
         Ok(SubResult::Unsub(success))
     }
+
+    /// Register the unsubscription callback for the new subscription on this connection
+    pub(crate) fn register_unsub(&mut self, handle: SubscriptionHandle) {
+        let cleanup = self.unsubs.insert(handle.id, handle.cleanup);
+        // If we have a duplicate subscription, drop the
+        // previous cleanup callback to prevent double-cleanup
+        if let Some(mut callback) = cleanup {
+            callback.0.take();
+        }
+    }
 }
 
 /// Bundles a connection's unique ID with its dedicated sender channel.