feat: support heartbeat transactions and proposer payouts

pyproject.toml

@@ -7,7 +7,7 @@ readme = "README.md"
 
 [tool.poetry.dependencies]
 python = "^3.12"
-py-algorand-sdk = "^2.6.1"
+py-algorand-sdk = "^2.7.0"
 algokit-utils = "^2.3.0"
 python-semantic-release = "^9.8.8"
 
@@ -154,13 +154,28 @@ prerelease_token = "beta"
 prerelease = true
 
 [tool.semantic_release.commit_parser_options]
-allowed_tags = ["build", "chore", "ci", "docs", "feat", "fix", "perf", "style", "refactor", "test"]
+allowed_tags = [
+    "build",
+    "chore",
+    "ci",
+    "docs",
+    "feat",
+    "fix",
+    "perf",
+    "style",
+    "refactor",
+    "test",
+]
 minor_tags = ["feat"]
 patch_tags = ["fix", "perf", "docs"]
 
 [tool.semantic_release.publish]
-dist_glob_patterns = ["dist/*", "stubs/dist/*"] # order here is important to ensure compiler wheel is published first
+dist_glob_patterns = [
+    "dist/*",
+    "stubs/dist/*",
+] # order here is important to ensure compiler wheel is published first
 upload_to_vcs_release = true
 
 [tool.semantic_release.remote.token]
-env = "GITHUB_TOKEN"
+env = "GITHUB_TOKEN"
+

src/algokit_subscriber/subscription.py

@@ -1084,9 +1084,9 @@ def filter_function(  # noqa: C901, PLR0912, PLR0915
     ) -> bool:
         t = txn["transaction"].dictify()  # type: ignore[no-untyped-call]
         created_app_id, created_asset_id, logs = (
-            txn["created_app_id"],
-            txn["created_asset_id"],
-            txn["logs"],
+            txn.get("created_app_id"),
+            txn.get("created_asset_id"),
+            txn.get("logs"),
         )
         result: bool = True
 

src/algokit_subscriber/transform.py

@@ -13,6 +13,7 @@
     KeyregTxn,
     PaymentTxn,
     StateProofTxn,
+    SuggestedParams,
     Transaction,
 )
 from typing_extensions import NotRequired  # noqa: UP035
@@ -70,46 +71,47 @@ def remove_nulls(obj: dict) -> dict:
     return obj
 
 
-# /**
-#  * Processes a block and returns all transactions from it, including inner transactions, with key information populated.
-#  * @param block An Algorand block
-#  * @returns An array of processed transactions from the block
-#  */
-# export function getBlockTransactions(block: Block): TransactionInBlock[] {
-#   let offset = 0
-#   const getOffset = () => offset++
+def get_transaction_from_block_payout(
+    block: Block, round_offset: int
+) -> TransactionInBlock:
+    """
+    Gets the synthetic transaction for the block payout as defined in the indexer
+
+    See /~https://github.com/algorand/indexer/blob/084577338ad4882f5797b3e1b30b84718ad40333/idb/postgres/internal/writer/write_txn.go?plain=1#L180-L202
+    """
+
+    pay = PaymentTxn(  # type: ignore[no-untyped-call]
+        sender=encode_address(block["fees"]),
+        receiver=encode_address(block.get("prp") or b""),
+        amt=block.get("pp", 0),
+        note=f"ProposerPayout for Round {block['rnd']}",
+        sp=SuggestedParams(  # type: ignore[no-untyped-call]
+            first=block["rnd"],
+            last=block["rnd"],
+            fee=0,
+            gh=base64.b64encode(block["gh"]).decode(),
+            gen=block["gen"],
+            flat_fee=True,
+            min_fee=0,
+        ),
+    )
+
+    txn: TransactionInBlock = {
+        "transaction": pay,
+        "round_offset": round_offset,
+        "round_number": block["rnd"],
+        "round_index": 0,
+        "block_transaction": {
+            "txn": pay.dictify()  # type: ignore[typeddict-item, misc, no-untyped-call]
+        },
+        "genesis_hash": block["gh"],
+        "genesis_id": block["gen"],
+        "round_timestamp": block["ts"],
+    }
+
+    return txn
 
 
-#   return (block.txns ?? []).flatMap((blockTransaction, roundIndex) => {
-#     let parentOffset = 0
-#     const getParentOffset = () => parentOffset++
-#     const parentData = extractTransactionFromBlockTransaction(blockTransaction, Buffer.from(block.gh), block.gen)
-#     return [
-#       {
-#         blockTransaction,
-#         block,
-#         roundOffset: getOffset(),
-#         roundIndex,
-#         roundNumber: block.rnd,
-#         roundTimestamp: block.ts,
-#         genesisId: block.gen,
-#         genesisHash: Buffer.from(block.gh),
-#         ...parentData,
-#       } as TransactionInBlock,
-#       ...(blockTransaction.dt?.itx ?? []).flatMap((innerTransaction) =>
-#         getBlockInnerTransactions(
-#           innerTransaction,
-#           block,
-#           blockTransaction,
-#           parentData.transaction.txID(),
-#           roundIndex,
-#           getOffset,
-#           getParentOffset,
-#         ),
-#       ),
-#     ]
-#   })
-# }
 def get_block_transactions(block: Block) -> list[TransactionInBlock]:
     txns: list[TransactionInBlock] = []
 
@@ -170,6 +172,9 @@ def get_parent_offset() -> int:
                 )
             )
 
+    if block.get("pp") and block.get("prp"):
+        txns.append(get_transaction_from_block_payout(block, get_offset()))
+
     return txns
 
 
@@ -417,13 +422,13 @@ def get_indexer_transaction_from_algod_transaction(  # noqa: C901
     filter_name: str | None = None,
 ) -> SubscribedTransaction:
     transaction = t["transaction"]
-    created_asset_id = t["created_asset_id"]
+    created_asset_id = t.get("created_asset_id")
     block_transaction = t["block_transaction"]
-    asset_close_amount = t["asset_close_amount"]
+    asset_close_amount = t.get("asset_close_amount")
     close_amount = t.get("close_amount")
     created_app_id = t.get("created_app_id")
     round_offset = t["round_offset"]
-    parent_offset = t["parent_offset"] or 0
+    parent_offset = t.get("parent_offset") or 0
     parent_transaction_id = t.get("parent_transaction_id")
     round_index = t["round_index"]
     round_number = t["round_number"]
@@ -541,7 +546,7 @@ def get_child_offset() -> int:
                 "extra-program-pages": transaction.extra_pages or None,
                 "foreign-apps": transaction.foreign_apps,
                 "foreign-assets": transaction.foreign_assets,
-                "accounts": transaction.accounts,
+                "accounts": transaction.accounts,  # type: ignore[typeddict-item]
             }
         elif isinstance(transaction, PaymentTxn):
             result["payment-transaction"] = {

src/algokit_subscriber/types/block.py

@@ -279,6 +279,34 @@ class Block(TypedDict):
     The transactions within the block.
     """
 
+    prp: NotRequired[None | bytes]
+    """
+    Proposer is the proposer of this block. Like the Seed, agreement adds
+    this after the block is assembled by the transaction pool, so that the same block can be prepared
+    for multiple participating accounts in the same node. Therefore, it can not be used
+    to influence block evaluation. Populated if proto.Payouts.Enabled
+    """
+
+    fc: NotRequired[None | int]
+    """
+    FeesCollected is the sum of all fees paid by transactions in this
+    block. Populated if proto.Payouts.Enabled
+    """
+
+    bi: NotRequired[None | int]
+    """
+    Bonus is the bonus incentive to be paid for proposing this block.  It
+	begins as a consensus parameter value, and decays periodically.
+    """
+
+    pp: NotRequired[None | int]
+    """
+    ProposerPayout is the amount that is moved from the FeeSink to
+    the Proposer in this block.  It is basically the
+    bonus + the payouts percent of FeesCollected, but may be zero'd by
+    proposer ineligibility.
+    """
+
 
 class BlockData(TypedDict):
     """

src/algokit_subscriber/types/transaction.py

@@ -49,6 +49,9 @@ class TransactionType(Enum):
     # State proof transaction
     stpf = "stpf"
 
+    # Heartbeat transaction
+    hb = "hb"
+
 
 class AlgodOnComplete(IntEnum):
     """

tests/filter_fixture.py

@@ -1,12 +1,24 @@
 import time
 
 import pytest
-from algokit_subscriber.types.subscription import TransactionSubscriptionResult
+from algokit_subscriber.types.subscription import (
+    SubscribedTransaction,
+    TransactionSubscriptionResult,
+)
 from algokit_utils.beta.algorand_client import AlgorandClient
 
 from .transactions import get_subscribed_transactions_for_test, send_x_transactions
 
 
+def filter_synthetic_transactions(t: SubscribedTransaction) -> bool:
+    return not (
+        t.get("tx-type") == "pay"
+        and t.get("fee", 0) == 0
+        and t.get("parent_transaction_id") is None
+        and t.get("group") is None
+    )
+
+
 @pytest.fixture()
 def filter_fixture() -> dict:
     localnet = AlgorandClient.default_local_net()
@@ -41,7 +53,7 @@ def subscribe_indexer(
                 )
                 time.sleep(1)
 
-        return get_subscribed_transactions_for_test(
+        result = get_subscribed_transactions_for_test(
             {
                 "max_rounds_to_sync": 1,
                 "sync_behaviour": "catchup-with-indexer",
@@ -53,6 +65,12 @@ def subscribe_indexer(
             algorand=localnet,
         )
 
+        # filter out txns with 0 fee, no parent, and no group
+        result["subscribed_transactions"] = list(
+            filter(filter_synthetic_transactions, result["subscribed_transactions"])
+        )
+        return result
+
     def subscribe_and_verify(
         txn_filter: dict, tx_id: str, arc28_events: list | None = None
     ) -> TransactionSubscriptionResult:

tests/test_balance_changes.py

@@ -1,6 +1,10 @@
-from typing import TYPE_CHECKING
-
-from algokit_subscriber.types.subscription import BalanceChangeRole
+from algokit_subscriber.subscriber import AlgorandSubscriber
+from algokit_subscriber.types.indexer import TransactionResult
+from algokit_subscriber.types.subscription import (
+    AlgorandSubscriberConfig,
+    BalanceChangeRole,
+)
+from algokit_utils.beta.algorand_client import AlgorandClient
 from algokit_utils.beta.composer import (
     AssetCreateParams,
     AssetDestroyParams,
@@ -9,9 +13,6 @@
     PayParams,
 )
 
-if TYPE_CHECKING:
-    from algokit_utils.beta.algorand_client import AlgorandClient
-
 from .accounts import generate_account
 from .filter_fixture import filter_fixture  # noqa: F401
 from .transactions import get_confirmations
@@ -857,3 +858,64 @@ def test_various_filters_on_axfers(filter_fixture: dict) -> None:  # noqa: PLR09
 
     for i, bc in enumerate(balance_changes):
         assert bc == expected_balance_changes[i]
+
+
+def test_block_payouts() -> None:
+    algorand = AlgorandClient.main_net()
+    payout_round = 46838092
+    watermark: int = payout_round - 1
+
+    def get_watermark() -> int | None:
+        return watermark
+
+    def set_watermark(n: int) -> None:
+        global watermark  # noqa: PLW0603
+        watermark = n
+
+    def synthetic_filter(txn: TransactionResult) -> bool:
+        return (
+            txn["sender"]
+            == "Y76M3MSY6DKBRHBL7C3NNDXGS5IIMQVQVUAB6MP4XEMMGVF2QWNPL226CA"
+        )
+
+    config: AlgorandSubscriberConfig = {
+        "filters": [
+            {
+                "name": "payout",
+                "filter": {"custom_filter": synthetic_filter},
+            }
+        ],
+        "max_rounds_to_sync": 1,
+        "max_indexer_rounds_to_sync": 1,
+        "sync_behaviour": "sync-oldest",
+        "watermark_persistence": {
+            "get": get_watermark,
+            "set": set_watermark,
+        },
+    }
+
+    algod_subscriber = AlgorandSubscriber(
+        config=config,
+        algod_client=algorand.client.algod,
+        indexer_client=algorand.client.indexer,
+    )
+    algod_result = algod_subscriber.poll_once()
+    assert len(algod_result["subscribed_transactions"]) == 1
+
+    watermark = payout_round - 1
+    config["sync_behaviour"] = "catchup-with-indexer"
+    indexer_subscriber = AlgorandSubscriber(
+        config=config,
+        algod_client=algorand.client.algod,
+        indexer_client=algorand.client.indexer,
+    )
+    indexer_result = indexer_subscriber.poll_once()
+
+    assert len(indexer_result["subscribed_transactions"]) == 1
+    assert (
+        algod_result["subscribed_transactions"][0]["id"]
+        == indexer_result["subscribed_transactions"][0]["id"]
+    )
+    assert algod_result["subscribed_transactions"][0].get(
+        "intra-round-offset"
+    ) == indexer_result["subscribed_transactions"][0].get("intra-round-offset")