From 5aee17233f238fe645624d3eeb050794def8655d Mon Sep 17 00:00:00 2001
From: drethereum <71602799+drethereum@users.noreply.github.com>
Date: Thu, 9 Nov 2023 12:48:51 -0700
Subject: [PATCH] add/beacon-blocks-retry (#749)

* retry for withdrawals

* test

* number seq

* simplify

* space

* updated test
---
 macros/tests/withdrawal_index_gaps.sql        | 37 ++++++++++++
 .../silver/silver__beacon_withdrawals.yml     |  4 ++
 .../streamline__beacon_blocks_realtime.sql    | 14 +++++
 .../beacon/retry/_missing_withdrawals.sql     | 58 +++++++++++++++++++
 4 files changed, 113 insertions(+)
 create mode 100644 macros/tests/withdrawal_index_gaps.sql
 create mode 100644 models/streamline/silver/beacon/retry/_missing_withdrawals.sql

diff --git a/macros/tests/withdrawal_index_gaps.sql b/macros/tests/withdrawal_index_gaps.sql
new file mode 100644
index 00000000..9348a624
--- /dev/null
+++ b/macros/tests/withdrawal_index_gaps.sql
@@ -0,0 +1,37 @@
+{% test withdrawal_index_gaps(
+    model,
+    column_name
+) %}
+WITH base AS (
+    SELECT
+        {{ column_name }} AS withdrawal_index,
+        LEAD(withdrawal_index) over (
+            ORDER BY
+                withdrawal_index
+        ) AS next_index,
+        slot_number AS start_slot_number,
+        LEAD(slot_number) over (
+            ORDER BY
+                withdrawal_index
+        ) AS end_slot_number
+    FROM
+        {{ model }}
+),
+gaps AS (
+    SELECT
+        withdrawal_index,
+        next_index,
+        withdrawal_index + 1 AS expected_index,
+        start_slot_number,
+        end_slot_number
+    FROM
+        base
+    WHERE
+        next_index IS NOT NULL
+        AND expected_index <> next_index
+)
+SELECT
+    next_index - withdrawal_index AS gaps
+FROM
+    gaps 
+{% endtest %}
diff --git a/models/beacon_chain/silver/silver__beacon_withdrawals.yml b/models/beacon_chain/silver/silver__beacon_withdrawals.yml
index 942ff756..05c399c1 100644
--- a/models/beacon_chain/silver/silver__beacon_withdrawals.yml
+++ b/models/beacon_chain/silver/silver__beacon_withdrawals.yml
@@ -53,6 +53,10 @@ models:
               column_type_list:
                 - NUMBER
                 - FLOAT
+          - withdrawal_index_gaps:
+              config:
+                severity: error
+                error_if: ">0"
       - name: VALIDATOR_INDEX
         tests:
           - not_null
diff --git a/models/streamline/silver/beacon/realtime/streamline__beacon_blocks_realtime.sql b/models/streamline/silver/beacon/realtime/streamline__beacon_blocks_realtime.sql
index 0f7a7443..ab8d4755 100644
--- a/models/streamline/silver/beacon/realtime/streamline__beacon_blocks_realtime.sql
+++ b/models/streamline/silver/beacon/realtime/streamline__beacon_blocks_realtime.sql
@@ -7,6 +7,8 @@
     tags = ['streamline_beacon_realtime']
 ) }}
 
+WITH to_do AS (
+
 SELECT
     {{ dbt_utils.generate_surrogate_key(
         ['slot_number']
@@ -25,3 +27,15 @@ FROM
     {{ ref("streamline__complete_beacon_blocks") }}
 WHERE
     slot_number > 5000000
+)
+
+SELECT 
+    id,
+    slot_number
+FROM to_do
+UNION
+SELECT
+    id,
+    slot_number
+FROM
+    {{ ref("_missing_withdrawals") }}
diff --git a/models/streamline/silver/beacon/retry/_missing_withdrawals.sql b/models/streamline/silver/beacon/retry/_missing_withdrawals.sql
new file mode 100644
index 00000000..47790601
--- /dev/null
+++ b/models/streamline/silver/beacon/retry/_missing_withdrawals.sql
@@ -0,0 +1,58 @@
+{{ config (
+    materialized = "ephemeral"
+) }}
+
+WITH base AS (
+
+    SELECT
+        INDEX AS withdrawal_index,
+        LEAD(withdrawal_index) over (
+            ORDER BY
+                withdrawal_index
+        ) AS next_index,
+        slot_number AS start_slot_number,
+        LEAD(slot_number) over (
+            ORDER BY
+                withdrawal_index
+        ) AS end_slot_number
+    FROM
+        {{ ref('silver__beacon_withdrawals') }}
+),
+gaps AS (
+    SELECT
+        withdrawal_index,
+        next_index,
+        withdrawal_index + 1 AS expected_index,
+        start_slot_number,
+        end_slot_number
+    FROM
+        base
+    WHERE
+        next_index IS NOT NULL
+        AND expected_index <> next_index
+),
+FINAL AS (
+    SELECT
+        withdrawal_index,
+        next_index,
+        expected_index,
+        start_slot_number,
+        end_slot_number,
+        start_slot_number + _id AS missing_slot_number
+    FROM
+        gaps
+        JOIN {{ ref('silver__number_sequence') }}
+        ON _id BETWEEN 1
+        AND (
+            end_slot_number - start_slot_number - 1
+        )
+    WHERE
+        missing_slot_number < end_slot_number
+)
+SELECT
+    DISTINCT missing_slot_number AS slot_number,
+    {{ dbt_utils.generate_surrogate_key(
+        ['slot_number']
+    ) }} AS id
+FROM
+    FINAL