Fix faucet

.github/workflows/deploy-faucet.yml

@@ -81,6 +81,7 @@ jobs:
             export FAUCET_RATE_LIMIT_WINDOW_SECONDS=86400
             export TXM_DB_PATH=/home/faucet/txm.sqlite
             export FAUCET_DB_PATH=/home/faucet/faucet.sqlite
+            export OTEL_EXPORTER_OTLP_ENDPOINT=http://148.113.212.211:4318/v1/traces
             EOF
               # cf. https://github.com/appleboy/drone-ssh/issues/175
               sed -i '/DRONE_SSH_PREV_COMMAND_EXIT_CODE/d' .env

.github/workflows/deploy-monitor-service.yml

@@ -66,8 +66,8 @@ jobs:
             export BLOCK_TIME=2
             export CHAIN_ID=216
             export RPC_URL=https://rpc.testnet.happy.tech/http
-            export MONITOR_ADDRESSES=0x10EBe5E4E8b4B5413D8e1f91A21cE4143B6bd8F5,0x3cBD2130C2D4D6aDAA9c9054360C29e00d99f0BA,0xBAc858b1AD51527F3c4A22f146246c9913e97cFd,0x84dcb507875af1786bb6623a625d3f9aae9fda4f,0xAE45fD410bf09f27DA574D3EF547567A479F4594,0x71E30C67d58015293f452468E4754b18bAFFd807,0xE55b09F1b78B72515ff1d1a0E3C14AD5D707fdE8,0x634de6fbFfE60EE6D1257f6be3E8AF4CfefEf697
-            export FUND_THRESHOLD=10000000000000000
+            export MONITOR_ADDRESSES=0x10EBe5E4E8b4B5413D8e1f91A21cE4143B6bd8F5,0x3cBD2130C2D4D6aDAA9c9054360C29e00d99f0BA,0xBAc858b1AD51527F3c4A22f146246c9913e97cFd,0x84dcb507875af1786bb6623a625d3f9aae9fda4f,0xAE45fD410bf09f27DA574D3EF547567A479F4594,0x71E30C67d58015293f452468E4754b18bAFFd807,0xE55b09F1b78B72515ff1d1a0E3C14AD5D707fdE8,0x634de6fbFfE60EE6D1257f6be3E8AF4CfefEf697,0xe8bD127b013600E5c6f864e5C07E918fa80BFF89
+            export FUND_THRESHOLD=100000000000000000
             export FUNDS_TO_SEND=10000000000000000
             export TXM_DB_PATH=/home/monitor_service/txm.sqlite
             export RPCS_TO_MONITOR=https://rpc.testnet.happy.tech/http,http://148.113.212.211:8545

Makefile

@@ -136,7 +136,7 @@ nuke: clean ## Removes build artifacts and dependencies
 	$(MAKE) remove-modules
 .PHONY: nuke
 
-test: sdk.test iframe.test ## Run tests
+test: sdk.test iframe.test txm.test ## Run tests
 .PHONY: test
 
 test.all: test contracts.test
@@ -266,17 +266,21 @@ sdk.test:
 	$(call forall_make , $(SDK_PKGS) , test)
 .PHONY: sdk.test
 
-sdk.check:
-	$(call forall_make , $(SDK_PKGS) , check)
-.PHONY: sdk.check
-
 iframe.test:
 	$(call forall_make , $(IFRAME_PKGS) , test)
 .PHONY: iframe.test
 
+txm.test:
+	cd packages/txm && make test
+.PHONY: iframe.test
+
 # ==================================================================================================
 # FORMATTING
 
+sdk.check:
+	$(call forall_make , $(SDK_PKGS) , check)
+.PHONY: sdk.check
+
 iframe.check:
 	$(call forall_make , $(IFRAME_PKGS) , check)
 .PHONY: iframe.check

apps/faucet/src/env.ts

@@ -24,6 +24,7 @@ const envSchema = z.object({
     TOKEN_AMOUNT: z.string().transform((s) => BigInt(s)),
     FAUCET_DB_PATH: z.string().trim(),
     FAUCET_RATE_LIMIT_WINDOW_SECONDS: z.string().transform((s) => Number(s)),
+    OTEL_EXPORTER_OTLP_ENDPOINT: z.string().trim().optional(),
 })
 
 const parsedEnv = envSchema.safeParse(process.env)

apps/faucet/src/services/faucet.ts

@@ -1,5 +1,6 @@
 import type { Address } from "@happy.tech/common"
 import { TransactionManager, TransactionStatus } from "@happy.tech/txm"
+import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-http"
 import { type Result, err, ok } from "neverthrow"
 import { env } from "../env"
 import { FaucetRateLimitError } from "../errors"
@@ -16,6 +17,15 @@ export class FaucetService {
             chainId: env.CHAIN_ID,
             blockTime: env.BLOCK_TIME,
             privateKey: env.PRIVATE_KEY,
+            traces: {
+                active: true,
+                spanExporter: env.OTEL_EXPORTER_OTLP_ENDPOINT
+                    ? new OTLPTraceExporter({
+                          url: env.OTEL_EXPORTER_OTLP_ENDPOINT,
+                      })
+                    : undefined,
+                serviceName: "faucet",
+            },
         })
         this.faucetUsageRepository = new FaucetUsageRepository()
     }
@@ -39,9 +49,6 @@ export class FaucetService {
             }
         }
 
-        const faucetUsage = FaucetUsage.create(address)
-        await this.faucetUsageRepository.save(faucetUsage)
-
         const tx = await this.txm.createTransaction({
             address,
             value: env.TOKEN_AMOUNT,
@@ -60,6 +67,9 @@ export class FaucetService {
             return err(new Error("Transaction failed"))
         }
 
+        const faucetUsage = FaucetUsage.create(address)
+        await this.faucetUsageRepository.save(faucetUsage)
+
         return ok(undefined)
     }
 }

packages/txm/lib/HookManager.ts

@@ -4,6 +4,15 @@ import type { Transaction } from "./Transaction.js"
 import type { AttemptSubmissionErrorCause } from "./TransactionSubmitter"
 import { TraceMethod } from "./telemetry/traces"
 
+// TODO
+//
+// TransactionSubmissionFailed is weird: it only triggers when the first attempt fails to submit in the Tx, but the
+// TxMonitor will still make further attempts (whose failure are not reported via hooks).
+//
+// TransactionSaveFailed has the same quirk, but there it makes sense as failing to save the transaction at the
+// collection stage means there won't be any further attempts, whereas further failures are not really user-relevant.
+// This is only used in tests, and probably this shouldn't be a user-exposed hook.
+
 export enum TxmHookType {
     All = "All",
     TransactionStatusChanged = "TransactionStatusChanged",

packages/txm/lib/Transaction.ts

@@ -13,7 +13,12 @@ import { TraceMethod } from "./telemetry/traces"
 
 export enum TransactionStatus {
     /**
-     * Default state for new transaction: the transaction is awaiting processing by TXM or has been submitted in the mempool and is waiting to be included in a block.
+     * Default state for new transaction: we're awaiting submission of the first attempt for the transaction.
+     */
+    NotAttempted = "NotAttempted",
+    /**
+     * At least one attempt to submit the transaction has been made — it might have hit the mempool and waiting for
+     * inclusion in a block, or it might have failed before that.
      */
     Pending = "Pending",
     /**
@@ -62,7 +67,11 @@ export enum TransactionCallDataFormat {
     Function = "Function",
 }
 
-export const NotFinalizedStatuses = [TransactionStatus.Pending, TransactionStatus.Cancelling]
+export const NotFinalizedStatuses = [
+    TransactionStatus.NotAttempted,
+    TransactionStatus.Pending,
+    TransactionStatus.Cancelling,
+]
 
 interface TransactionConstructorBaseConfig {
     /**
@@ -181,7 +190,7 @@ export class Transaction {
         this.address = config.address
         this.value = config.value ?? 0n
         this.deadline = config.deadline
-        this.status = config.status ?? TransactionStatus.Pending
+        this.status = config.status ?? TransactionStatus.NotAttempted
         this.attempts = config.attempts ?? []
         this.collectionBlock = config.collectionBlock
         this.createdAt = config.createdAt ?? new Date()

packages/txm/lib/TransactionCollector.ts

@@ -95,7 +95,7 @@ export class TransactionCollector {
                 })
 
                 if (transaction.status === TransactionStatus.Interrupted) {
-                    transaction.changeStatus(TransactionStatus.Pending)
+                    transaction.changeStatus(TransactionStatus.NotAttempted)
                 }
 
                 const submissionResult = await this.txmgr.transactionSubmitter.submitNewAttempt(transaction, {
@@ -105,6 +105,9 @@ export class TransactionCollector {
                     maxPriorityFeePerGas,
                 })
 
+                // Only after submitting the initial attempt to avoid concurrent attempts here & in the TxMonitor.
+                transaction.changeStatus(TransactionStatus.Pending)
+
                 if (submissionResult.isErr()) {
                     eventBus.emit(Topics.TransactionSubmissionFailed, {
                         transaction,

packages/txm/lib/TransactionManager.ts

@@ -252,6 +252,12 @@ export type TransactionManagerConfig = {
          * Defaults to a console span exporter.
          */
         spanExporter?: SpanExporter
+
+        /**
+         * The service name to use for the traces.
+         * Defaults to "txm".
+         */
+        serviceName?: string
     }
 }
 
@@ -307,6 +313,7 @@ export class TransactionManager {
             userMetricReader: _config.metrics?.metricReader,
             tracesActive: _config.traces?.active ?? false,
             userTraceExporter: _config.traces?.spanExporter,
+            serviceName: _config.traces?.serviceName,
         })
 
         this.collectors = []

packages/txm/lib/TransactionRepository.ts

@@ -3,7 +3,7 @@ import type { UUID } from "@happy.tech/common"
 import { SpanStatusCode, context, trace } from "@opentelemetry/api"
 import { type Result, ResultAsync, err, ok } from "neverthrow"
 import { Topics, eventBus } from "./EventBus.js"
-import { NotFinalizedStatuses, Transaction } from "./Transaction.js"
+import { NotFinalizedStatuses, Transaction, TransactionStatus } from "./Transaction.js"
 import type { TransactionManager } from "./TransactionManager.js"
 import { db } from "./db/driver.js"
 import { TxmMetrics } from "./telemetry/metrics"
@@ -42,9 +42,11 @@ export class TransactionRepository {
         }
     }
 
-    @TraceMethod("txm.transaction-repository.get-not-finalized-transactions-older-than")
-    getNotFinalizedTransactionsOlderThan(blockNumber: bigint): Transaction[] {
-        return this.notFinalizedTransactions.filter((t) => t.collectionBlock && t.collectionBlock < blockNumber)
+    @TraceMethod("txm.transaction-repository.get-in-flight-transactions-older-than")
+    getInFlightTransactionsOlderThan(blockNumber: bigint): Transaction[] {
+        return this.notFinalizedTransactions.filter(
+            (t) => t.status !== TransactionStatus.NotAttempted && t.collectionBlock! < blockNumber,
+        )
     }
 
     @TraceMethod("txm.transaction-repository.get-transaction")

packages/txm/lib/TxMonitor.ts

@@ -81,6 +81,7 @@ export class TxMonitor {
     @TraceMethod("txm.tx-monitor.handle-new-block")
     private async handleNewBlock(block: LatestBlock) {
         const span = trace.getSpan(context.active())!
+        const txRepository = this.transactionManager.transactionRepository
 
         span.addEvent("txm.tx-monitor.handle-new-block.started", {
             blockNumber: Number(block.number),
@@ -93,9 +94,7 @@ export class TxMonitor {
             return
         }
 
-        const transactions = this.transactionManager.transactionRepository.getNotFinalizedTransactionsOlderThan(
-            block.number,
-        )
+        const transactions = txRepository.getInFlightTransactionsOlderThan(block.number)
 
         for (const transaction of transactions) {
             span.addEvent("txm.tx-monitor.handle-new-block.monitoring-transaction", {
@@ -228,10 +227,7 @@ export class TxMonitor {
 
         await Promise.all(promises)
 
-        const result = await ResultAsync.fromPromise(
-            this.transactionManager.transactionRepository.saveTransactions(transactions),
-            unknownToError,
-        )
+        const result = await ResultAsync.fromPromise(txRepository.saveTransactions(transactions), unknownToError)
 
         if (result.isErr()) {
             logger.error("Error flushing transactions in onNewBlock")

packages/txm/lib/telemetry/instrumentation.ts

@@ -4,27 +4,30 @@ import type { MetricReader } from "@opentelemetry/sdk-metrics"
 import { NodeSDK } from "@opentelemetry/sdk-node"
 import { ConsoleSpanExporter, type SpanExporter } from "@opentelemetry/sdk-trace-node"
 import { ATTR_SERVICE_NAME, ATTR_SERVICE_VERSION } from "@opentelemetry/semantic-conventions"
-
-const resource = Resource.default().merge(
-    new Resource({
-        [ATTR_SERVICE_NAME]: "txm",
-        [ATTR_SERVICE_VERSION]: "0.1.0",
-    }),
-)
+import { version } from "../../package.json"
 
 export function initializeTelemetry({
     metricsActive,
     prometheusPort,
     userMetricReader,
     tracesActive,
     userTraceExporter,
+    serviceName,
 }: {
     metricsActive: boolean
     prometheusPort: number
     userMetricReader?: MetricReader
     userTraceExporter?: SpanExporter
     tracesActive?: boolean
+    serviceName?: string
 }): void {
+    const resource = Resource.default().merge(
+        new Resource({
+            [ATTR_SERVICE_NAME]: serviceName ?? "txm",
+            [ATTR_SERVICE_VERSION]: version,
+        }),
+    )
+
     let metricReader: MetricReader | undefined
     if (metricsActive) {
         metricReader = userMetricReader || new PrometheusExporter({ port: prometheusPort })