Merge branch 'main' into satya/fungible-concurrenty-supply

.github/workflows/lint.yaml

@@ -31,6 +31,7 @@ jobs:
       - uses: actions/checkout@v3
       - name: Install deps and run linter
         run: |
+          sudo apt update && sudo apt install libdw-dev
           cargo install cargo-sort
           rustup update
           rustup toolchain install nightly

README.md

@@ -24,3 +24,6 @@ This guide will get you started with creating an Aptos indexer with custom parsi
 
 ## [Aptos Indexer GRPC Release Notes](https://github.com/aptos-labs/aptos-core/blob/main/ecosystem/indexer-grpc/release_notes.md)
 
+
+> [!WARNING]  
+> The typescript implementation is known to get stuck when there are lots of data to process. The issue is with the GRPC client and we haven't had a chance to optimize. Please proceed with caution.

hasura-api/metadata-json/unified.json

@@ -1,5 +1,5 @@
 {
-  "resource_version": 307,
+  "resource_version": 316,
   "metadata": {
     "version": 3,
     "sources": [
@@ -295,6 +295,23 @@
               "name": "account_transactions",
               "schema": "public"
             },
+            "object_relationships": [
+              {
+                "name": "user_transaction",
+                "using": {
+                  "manual_configuration": {
+                    "column_mapping": {
+                      "transaction_version": "version"
+                    },
+                    "insertion_order": null,
+                    "remote_table": {
+                      "name": "user_transactions",
+                      "schema": "public"
+                    }
+                  }
+                }
+              }
+            ],
             "array_relationships": [
               {
                 "name": "coin_activities",
@@ -879,17 +896,19 @@
                 "role": "anonymous",
                 "permission": {
                   "columns": [
-                    "last_transaction_version",
+                    "domain",
+                    "domain_expiration_timestamp",
+                    "domain_with_suffix",
+                    "expiration_timestamp",
                     "is_active",
                     "is_primary",
-                    "domain",
+                    "last_transaction_version",
                     "owner_address",
                     "registered_address",
                     "subdomain",
+                    "subdomain_expiration_policy",
                     "token_name",
-                    "token_standard",
-                    "domain_with_suffix",
-                    "expiration_timestamp"
+                    "token_standard"
                   ],
                   "filter": {},
                   "limit": 100,
@@ -2483,7 +2502,7 @@
           "GET"
         ],
         "name": "Latest Processor Status",
-        "url": "get_lastest_processor_status"
+        "url": "get_latest_processor_status"
       }
     ],
     "api_limits": {

python/utils/worker.py

@@ -201,8 +201,24 @@ def producer(
                 extra={
                     "processor_name": processor_name,
                     "stream_address": indexer_grpc_data_service_address,
+                    "error": str(e),
+                    "next_version_to_fetch": next_version_to_fetch,
+                    "ending_version": ending_version,
                 },
             )
+            # Datastream error can happen when we fail to deserialize deeply nested types.
+            # Skip the batch, log the error, and continue processing.
+            is_success = True
+            next_version_to_fetch += 1
+            response_stream = get_grpc_stream(
+                indexer_grpc_data_service_address,
+                indexer_grpc_data_stream_api_key,
+                indexer_grpc_http2_ping_interval,
+                indexer_grpc_http2_ping_timeout,
+                next_version_to_fetch,
+                ending_version,
+                processor_name,
+            )
 
         # Check if we're at the end of the stream
         reached_ending_version = (
@@ -369,7 +385,8 @@ async def consumer_impl(
                         "service_type": PROCESSOR_SERVICE_TYPE,
                     },
                 )
-                os._exit(1)
+                # Gaps are possible because we skipped versions
+                # os._exit(1)
             last_fetched_version = transactions[-1].version
             transaction_batches.append(transactions)
 
@@ -415,7 +432,8 @@ async def consumer_impl(
                             "service_type": PROCESSOR_SERVICE_TYPE,
                         },
                     )
-                    os._exit(1)
+                    # Gaps are possible because we skip versions
+                    # os._exit(1)
                 prev_start = result.start_version
                 prev_end = result.end_version