Change importer to work with binary (#779)

* add timing around get_object and parse csv

* reduce terminationGracePeriodSeconds

* bump image

* convert to binary

* treat id as u32

* change folder name

* deal only with specified folder

* use 8k

* set parallelism to 8

* run 2k with 16 par

* use 8 par

* set to 2k + 32 par

* switch to bytes

* rebase

* bump image

* test config

* test config

* generate chunk names based on timestamp file

* use 1k chunk in stage 0

* use 512 chunk in stage 0

* fix test

* remove left overs

* fmt

* use IRIS_CODE_LENGTH and MASK_CODE_LENGTH instead of const

.github/workflows/temp-branch-build-and-push.yaml

@@ -0,0 +1,47 @@
+name: Branch - Build and push docker image
+
+on:
+  push:
+    branches:
+      - "ertugrul/change-to-binary"
+
+concurrency:
+  group: '${{ github.workflow }} @ ${{ github.event.pull_request.head.label || github.head_ref || github.ref }}'
+  cancel-in-progress: true
+
+env:
+  REGISTRY: ghcr.io
+  IMAGE_NAME: ${{ github.repository }}
+
+jobs:
+  docker:
+    runs-on:
+      labels: ubuntu-22.04-64core
+    permissions:
+      packages: write
+      contents: read
+      attestations: write
+      id-token: write
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+      - name: Set up QEMU
+        uses: docker/setup-qemu-action@v3
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v3
+      - name: Log in to the Container registry
+        uses: docker/login-action@v3
+        with:
+          registry: ${{ env.REGISTRY }}
+          username: ${{ github.repository_owner }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+      - name: Build and Push
+        uses: docker/build-push-action@v6
+        with:
+          context: .
+          push: true
+          tags: |
+            ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:${{ github.sha }}
+          platforms: linux/amd64
+          cache-from: type=gha
+          cache-to: type=gha,mode=max

deploy/stage/common-values-iris-mpc.yaml

@@ -1,4 +1,4 @@
-image: "ghcr.io/worldcoin/iris-mpc:v0.12.3"
+image: "ghcr.io/worldcoin/iris-mpc:v0.12.4"
 
 environment: stage
 replicaCount: 1
@@ -82,4 +82,4 @@ preStop:
   sleepPeriod: 10
 
 # terminationGracePeriodSeconds specifies the grace time between SIGTERM and SIGKILL
-terminationGracePeriodSeconds: 180 # 3x SMPC__PROCESSING_TIMEOUT_SECS
+terminationGracePeriodSeconds: 20

deploy/stage/smpcv2-0-stage/values-iris-mpc.yaml

@@ -74,6 +74,12 @@ env:
   - name: SMPC__DB_CHUNKS_BUCKET_NAME
     value: "iris-mpc-db-exporter-store-node-0-stage-eu-north-1"
 
+  - name: SMPC__DB_CHUNKS_FOLDER_NAME
+    value: "binary_output_2k"
+
+  - name: SMPC__LOAD_CHUNKS_PARALLELISM
+    value: "32"
+
   - name: SMPC__CLEAR_DB_BEFORE_INIT
     value: "true"
 

deploy/stage/smpcv2-1-stage/values-iris-mpc.yaml

@@ -74,6 +74,12 @@ env:
   - name: SMPC__DB_CHUNKS_BUCKET_NAME
     value: "iris-mpc-db-exporter-store-node-1-stage-eu-north-1"
 
+  - name: SMPC__DB_CHUNKS_FOLDER_NAME
+    value: "binary_output_2k"
+
+  - name: SMPC__LOAD_CHUNKS_PARALLELISM
+    value: "32"
+
   - name: SMPC__CLEAR_DB_BEFORE_INIT
     value: "true"
 

deploy/stage/smpcv2-2-stage/values-iris-mpc.yaml

@@ -74,6 +74,12 @@ env:
   - name: SMPC__DB_CHUNKS_BUCKET_NAME
     value: "iris-mpc-db-exporter-store-node-2-stage-eu-north-1"
 
+  - name: SMPC__DB_CHUNKS_FOLDER_NAME
+    value: "binary_output_2k"
+
+  - name: SMPC__LOAD_CHUNKS_PARALLELISM
+    value: "32"
+
   - name: SMPC__CLEAR_DB_BEFORE_INIT
     value: "true"
 

iris-mpc-common/src/config/mod.rs

@@ -102,6 +102,9 @@ pub struct Config {
     /// updated during the DB export to S3
     #[serde(default = "default_db_load_safety_overlap_seconds")]
     pub db_load_safety_overlap_seconds: i64,
+
+    #[serde(default)]
+    pub db_chunks_folder_name: String,
 }
 
 fn default_load_chunks_parallelism() -> usize {

iris-mpc-store/src/lib.rs

@@ -1,3 +1,5 @@
+#![feature(int_roundings)]
+
 mod s3_importer;
 
 use bytemuck::cast_slice;
@@ -10,6 +12,7 @@ use iris_mpc_common::{
     config::Config,
     galois_engine::degree4::{GaloisRingIrisCodeShare, GaloisRingTrimmedMaskCodeShare},
     iris_db::iris::IrisCode,
+    IRIS_CODE_LENGTH, MASK_CODE_LENGTH,
 };
 use rand::{rngs::StdRng, Rng, SeedableRng};
 pub use s3_importer::{fetch_and_parse_chunks, last_snapshot_timestamp, ObjectStore, S3Store};
@@ -74,6 +77,43 @@ impl StoredIris {
     pub fn id(&self) -> i64 {
         self.id
     }
+
+    pub fn from_bytes(bytes: &[u8]) -> Result<Self, eyre::Error> {
+        let mut cursor = 0;
+
+        // Helper closure to extract a slice of a given size
+        let extract_slice =
+            |bytes: &[u8], cursor: &mut usize, size: usize| -> Result<Vec<u8>, eyre::Error> {
+                if *cursor + size > bytes.len() {
+                    return Err(eyre!("Exceeded total bytes while extracting slice",));
+                }
+                let slice = &bytes[*cursor..*cursor + size];
+                *cursor += size;
+                Ok(slice.to_vec())
+            };
+
+        // Parse `id` (i64)
+        let id_bytes = extract_slice(bytes, &mut cursor, 4)?;
+        let id = u32::from_be_bytes(
+            id_bytes
+                .try_into()
+                .map_err(|_| eyre!("Failed to convert id bytes to i64"))?,
+        ) as i64;
+
+        // parse codes and masks
+        let left_code = extract_slice(bytes, &mut cursor, IRIS_CODE_LENGTH * size_of::<u16>())?;
+        let left_mask = extract_slice(bytes, &mut cursor, MASK_CODE_LENGTH * size_of::<u16>())?;
+        let right_code = extract_slice(bytes, &mut cursor, IRIS_CODE_LENGTH * size_of::<u16>())?;
+        let right_mask = extract_slice(bytes, &mut cursor, MASK_CODE_LENGTH * size_of::<u16>())?;
+
+        Ok(StoredIris {
+            id,
+            left_code,
+            left_mask,
+            right_code,
+            right_mask,
+        })
+    }
 }
 
 #[derive(Clone)]