feat(backend): connect the inactivity watchdog to on-chain Soroban payout execution (#1045)

* fix(backend): repair the migration chain so it applies from scratch

`sqlx::migrate!()` aborted on the first failing migration, so on an empty
database nothing after the core tables was ever created — including the
inactivity watchdog's own `status` / `inactivity_deadline_at` columns, which
the watchdog then queried on every sweep.

Three separate breakages:

- `20260624000000_add_inactivity_watchdog_fields` derived
  `inactivity_deadline_at` by adding an INTERVAL to `last_ping`, but
  `last_ping` is a BIGINT of Unix seconds (core tables migration, and the API
  binds it as an i64), so the statement failed with "operator does not exist:
  bigint + interval". Derive the deadline with `to_timestamp()` instead, and
  drop the no-op attempt to re-add `last_ping` as a timestamp.

- `20260628000000` was used by two migrations (`add_kyc_records` and
  `create_admins`); sqlx keys migrations by version and rejected the
  duplicate with VersionMismatch. Renumber `create_admins` to
  `20260628000001`, preserving the original order.

- `20260729000000_create_plans_beneficiaries_payouts` re-created `plans` and
  `beneficiaries`, which the core tables migration already creates, failing
  with "relation \"plans\" already exists". Only `payout_logs` was new, and
  nothing in the codebase references the duplicate definitions' columns, so
  keep just that table and point it at the existing `beneficiaries`.

Verified by applying the full chain to an empty PostgreSQL database through
`DbManager::run_migrations`.

* feat(backend): connect the inactivity watchdog to on-chain Soroban payout execution

Closes #1039

The inactivity watchdog only ever flipped an expired plan's status to
TRIGGERED in PostgreSQL and enqueued a `plan.triggered` webhook; the funds
were never unlocked on-chain. It now calls `trigger_inheritance` on the
Soroban inheritance contract and only records the plan as triggered once the
contract confirms it.

stellar_submit.rs — StellarSubmitClient learns to invoke Soroban contracts:

- `SorobanConfig::from_env()` reads SOROBAN_RPC_URL, INHERITANCE_CONTRACT_ID,
  STELLAR_SIGNER_SECRET (plus optional passphrase and poll settings), and
  `with_soroban()` validates the strkeys up front so a misconfigured signer
  fails at startup rather than on the first expired plan.
- `invoke_contract()` builds an InvokeHostFunction transaction, simulates it
  over JSON-RPC to pick up the footprint, resource fee and auth entries,
  signs it for the configured network, submits it and polls `getTransaction`
  until it lands or the timeout expires.
- `contract_events()` / `find_event()` / `event_u64_field()` decode the
  emitted events out of TransactionMeta V3 and V4.

inactivity_watchdog.rs — the sweep is now two-phase:

- Expired plans are claimed into a new TRIGGERING status under the existing
  advisory lock, then triggered on-chain outside the transaction. A plan only
  becomes TRIGGERED after the contract's INHERIT/TRIGGER event is found and
  its plan_id matches the one we asked for; anything else lands in
  TRIGGER_FAILED with the reason recorded.
- Failed submissions are retried with exponential backoff
  (INACTIVITY_WATCHDOG_ONCHAIN_MAX_ATTEMPTS / _BACKOFF_MS / _MAX_BACKOFF_MS).
  Only transport and inclusion failures are retried — a contract-level
  rejection will be rejected identically next time.
- Alerting: failures log at error level, increment the new
  `inheritx_watchdog_onchain_triggers_total{outcome="failure"}` counter, and
  enqueue a `plan.trigger_failed` webhook. `plan.triggered` now carries the
  transaction hash.
- A plan left in TRIGGERING by a crashed worker is re-claimed once it goes
  stale (INACTIVITY_WATCHDOG_TRIGGER_STALE_AFTER_SECS, default 15 minutes).
- With no Soroban configuration the watchdog keeps its previous behaviour and
  says so loudly at startup.

Supporting changes:

- New migration adds `onchain_plan_id` (the contract's u64 plan id, unique
  where present), plus `trigger_tx_hash`, `trigger_attempts`,
  `trigger_started_at` and `last_trigger_error`.
- `POST /api/plans` accepts an optional `onchain_plan_id` and it is returned
  on plan responses, so a plan can actually be linked to its contract plan.
- main.rs builds one Stellar client and shares it between the API and the
  watchdog.

Tests: unit coverage for the signing payload, event decoding and matching,
config validation, retry classification and backoff; plus database-backed
sweep tests. CI's PostgreSQL service now has a health check and the tests get
WATCHDOG_TEST_DATABASE_URL so those run (deliberately not DATABASE_URL, which
other tests expect to be unreachable).

* fix(backend): only count a watchdog trigger as an on-chain success when one happened

`inheritx_watchdog_onchain_triggers_total{outcome="success"}` was also
incremented on the fallback path taken when no Soroban signer is configured,
where the watchdog never contacts the chain. Gate it on an actual transaction
hash so the series only reflects real submissions.

Author: Yunusabdul38

.github/workflows/backend.yml

@@ -28,6 +28,11 @@ jobs:
           POSTGRES_PASSWORD: password
         ports:
           - 5432:5432
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
 
     steps:
       - name: Checkout code
@@ -52,6 +57,12 @@ jobs:
 
       - name: Run tests
         run: cargo test
+        env:
+          # Opts the inactivity watchdog into the database-backed tests, which
+          # also verify the migration chain applies from scratch. Deliberately
+          # not DATABASE_URL: other tests assert behaviour when the database is
+          # unreachable.
+          WATCHDOG_TEST_DATABASE_URL: postgres://postgres:password@localhost:5432/test
 
       - name: Build
         run: cargo build --release

backend/migrations/20260624000000_add_inactivity_watchdog_fields.sql

@@ -1,8 +1,13 @@
 -- Issue #820: persist proof-of-life inactivity timers for the watchdog worker.
+--
+-- `plans.last_ping` is a BIGINT of Unix seconds (see the core tables
+-- migration) and the API binds it as such, so the deadline is derived with
+-- to_timestamp() rather than by adding an INTERVAL to a timestamp.
+
+ALTER TABLE plans
+    ADD COLUMN IF NOT EXISTS status VARCHAR(32) NOT NULL DEFAULT 'ACTIVE';
 
 ALTER TABLE plans
-    ADD COLUMN IF NOT EXISTS status VARCHAR(32) NOT NULL DEFAULT 'ACTIVE',
-    ADD COLUMN IF NOT EXISTS last_ping TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(),
     ADD COLUMN IF NOT EXISTS grace_period_seconds BIGINT NOT NULL DEFAULT 7776000;
 
 ALTER TABLE plans
@@ -15,9 +20,7 @@ ALTER TABLE plans
 
 ALTER TABLE plans
     ADD COLUMN IF NOT EXISTS inactivity_deadline_at TIMESTAMP WITH TIME ZONE
-    GENERATED ALWAYS AS (
-        last_ping + (grace_period_seconds::double precision * INTERVAL '1 second')
-    ) STORED;
+    GENERATED ALWAYS AS (to_timestamp(last_ping + grace_period_seconds)) STORED;
 
 CREATE INDEX IF NOT EXISTS idx_plans_inactivity_deadline_claimable
     ON plans (inactivity_deadline_at)

…ns/20260628000000_create_admins.down.sql …ns/20260628000001_create_admins.down.sqlbackend/migrations/20260628000000_create_admins.down.sql renamed to backend/migrations/20260628000001_create_admins.down.sql backend/migrations/20260628000000_create_admins.down.sql renamed to backend/migrations/20260628000001_create_admins.down.sql

@@ -1,3 +1,2 @@
-DROP TABLE payout_logs;
-DROP TABLE beneficiaries;
-DROP TABLE plans;
+DROP INDEX IF EXISTS payout_logs_beneficiary_id_idx;
+DROP TABLE IF EXISTS payout_logs;

…ions/20260628000000_create_admins.up.sql …ions/20260628000001_create_admins.up.sqlbackend/migrations/20260628000000_create_admins.up.sql renamed to backend/migrations/20260628000001_create_admins.up.sql backend/migrations/20260628000000_create_admins.up.sql renamed to backend/migrations/20260628000001_create_admins.up.sql

@@ -1,30 +1,16 @@
-CREATE TABLE plans (
-    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
-    name TEXT NOT NULL,
-    description TEXT,
-    apy_rate_bps INT NOT NULL,
-    min_amount NUMERIC(19, 4) NOT NULL,
-    max_amount NUMERIC(19, 4),
-    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
-    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
-);
-
-CREATE TABLE beneficiaries (
-    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
-    owner_wallet_address TEXT NOT NULL,
-    beneficiary_wallet_address TEXT NOT NULL,
-    share_percentage NUMERIC(5, 2) NOT NULL,
-    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
-    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
-);
+-- Issue #1017: payout audit log.
+--
+-- `plans` and `beneficiaries` are already created by the core tables
+-- migration, and re-creating them here aborted the whole migration chain
+-- ("relation \"plans\" already exists"), so this migration now only adds the
+-- table that was genuinely new. It references the existing `beneficiaries`.
 
-CREATE TABLE payout_logs (
+CREATE TABLE IF NOT EXISTS payout_logs (
     id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
-    beneficiary_id UUID NOT NULL REFERENCES beneficiaries(id),
+    beneficiary_id UUID NOT NULL REFERENCES beneficiaries (id) ON DELETE CASCADE,
     amount NUMERIC(19, 4) NOT NULL,
     payout_date TIMESTAMPTZ NOT NULL DEFAULT NOW(),
     transaction_hash TEXT NOT NULL
 );
 
-CREATE INDEX beneficiaries_owner_wallet_idx ON beneficiaries (owner_wallet_address);
-CREATE INDEX payout_logs_beneficiary_id_idx ON payout_logs (beneficiary_id);
+CREATE INDEX IF NOT EXISTS payout_logs_beneficiary_id_idx ON payout_logs (beneficiary_id);

backend/migrations/20260729000000_create_plans_beneficiaries_payouts.down.sql

@@ -0,0 +1,12 @@
+DROP INDEX IF EXISTS idx_plans_trigger_in_flight;
+DROP INDEX IF EXISTS plans_onchain_plan_id_key;
+
+ALTER TABLE plans
+    DROP CONSTRAINT IF EXISTS plans_onchain_plan_id_non_negative;
+
+ALTER TABLE plans
+    DROP COLUMN IF EXISTS last_trigger_error,
+    DROP COLUMN IF EXISTS trigger_started_at,
+    DROP COLUMN IF EXISTS trigger_attempts,
+    DROP COLUMN IF EXISTS trigger_tx_hash,
+    DROP COLUMN IF EXISTS onchain_plan_id;

backend/migrations/20260729000000_create_plans_beneficiaries_payouts.up.sql

@@ -0,0 +1,29 @@
+-- Issue #1039: the inactivity watchdog now executes the payout on-chain before
+-- flipping a plan to TRIGGERED, so it needs the Soroban plan id to call
+-- `trigger_inheritance` with, plus somewhere to record the attempt.
+
+ALTER TABLE plans
+    ADD COLUMN IF NOT EXISTS onchain_plan_id BIGINT,
+    ADD COLUMN IF NOT EXISTS trigger_tx_hash TEXT,
+    ADD COLUMN IF NOT EXISTS trigger_attempts INTEGER NOT NULL DEFAULT 0,
+    ADD COLUMN IF NOT EXISTS trigger_started_at TIMESTAMP WITH TIME ZONE,
+    ADD COLUMN IF NOT EXISTS last_trigger_error TEXT;
+
+ALTER TABLE plans
+    ADD CONSTRAINT plans_onchain_plan_id_non_negative
+    CHECK (onchain_plan_id IS NULL OR onchain_plan_id >= 0)
+    NOT VALID;
+
+ALTER TABLE plans
+    VALIDATE CONSTRAINT plans_onchain_plan_id_non_negative;
+
+-- One database plan per on-chain plan: triggering the same contract plan from
+-- two rows would double-submit the payout.
+CREATE UNIQUE INDEX IF NOT EXISTS plans_onchain_plan_id_key
+    ON plans (onchain_plan_id)
+    WHERE onchain_plan_id IS NOT NULL;
+
+-- Lets the watchdog cheaply find submissions left in flight by a crashed worker.
+CREATE INDEX IF NOT EXISTS idx_plans_trigger_in_flight
+    ON plans (trigger_started_at)
+    WHERE status = 'TRIGGERING';

backend/migrations/20260820000000_add_onchain_trigger_fields.down.sql

@@ -53,6 +53,10 @@ pub struct Plan {
     pub earn_yield: bool,
     pub yield_rate_bps: u32,
     pub is_active: bool,
+    /// Identifier of this plan inside the Soroban inheritance contract.
+    /// Required for the inactivity watchdog to trigger the payout on-chain;
+    /// plans created without one can only be triggered manually.
+    pub onchain_plan_id: Option<u64>,
 }
 
 pub struct AppState {
@@ -312,6 +316,7 @@ pub struct PlanRow {
     pub yield_rate_bps: i32,
     pub accrued_yield: rust_decimal::Decimal,
     pub created_at: chrono::DateTime<chrono::Utc>,
+    pub onchain_plan_id: Option<i64>,
 }
 
 #[derive(Debug, Clone, Serialize, sqlx::FromRow)]
@@ -345,6 +350,7 @@ pub struct PlanResponse {
     pub yield_rate_bps: i32,
     pub accrued_yield: f64,
     pub created_at: chrono::DateTime<chrono::Utc>,
+    pub onchain_plan_id: Option<i64>,
     pub beneficiaries: Vec<BeneficiaryResponse>,
 }
 
@@ -432,6 +438,7 @@ fn plan_row_to_response(row: PlanRow, beneficiaries: Vec<BeneficiaryResponse>) -
         yield_rate_bps: row.yield_rate_bps,
         accrued_yield,
         created_at: row.created_at,
+        onchain_plan_id: row.onchain_plan_id,
         beneficiaries,
     }
 }
@@ -622,9 +629,10 @@ async fn create_plan(
             accrued_yield,
             last_ping,
             is_active,
-            status
-        ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
-        RETURNING id, owner_address, token_address, amount, grace_period, grace_period_seconds, earn_yield, last_ping, is_active, status, yield_rate_bps, accrued_yield, created_at
+            status,
+            onchain_plan_id
+        ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
+        RETURNING id, owner_address, token_address, amount, grace_period, grace_period_seconds, earn_yield, last_ping, is_active, status, yield_rate_bps, accrued_yield, created_at, onchain_plan_id
         "#
     )
     .bind(&payload.owner)
@@ -638,6 +646,7 @@ async fn create_plan(
     .bind(payload.last_ping)
     .bind(payload.is_active)
     .bind("ACTIVE")
+    .bind(payload.onchain_plan_id.map(|id| id as i64))
     .fetch_one(&mut *tx)
     .await {
         Ok(row) => row,
@@ -715,6 +724,7 @@ async fn create_plan(
         yield_rate_bps: plan_row.yield_rate_bps,
         accrued_yield: 0.0, // No yield accrued at creation
         created_at: plan_row.created_at,
+        onchain_plan_id: plan_row.onchain_plan_id,
         beneficiaries: inserted_beneficiaries,
     };
 
@@ -787,7 +797,7 @@ async fn update_plan(
 
     // 3. Check if plan exists
     let _plan_row = match sqlx::query_as::<_, PlanRow>(
-        "SELECT id, owner_address, token_address, amount, grace_period, grace_period_seconds, earn_yield, last_ping, is_active, status, yield_rate_bps, accrued_yield, created_at FROM plans WHERE id = $1"
+        "SELECT id, owner_address, token_address, amount, grace_period, grace_period_seconds, earn_yield, last_ping, is_active, status, yield_rate_bps, accrued_yield, created_at, onchain_plan_id FROM plans WHERE id = $1"
     )
     .bind(plan_id)
     .fetch_optional(&mut *tx)
@@ -903,7 +913,7 @@ async fn update_plan(
 
     // 7. Fetch updated plan with beneficiaries
     let updated_plan_row = match sqlx::query_as::<_, PlanRow>(
-        "SELECT id, owner_address, token_address, amount, grace_period, grace_period_seconds, earn_yield, last_ping, is_active, status, yield_rate_bps, accrued_yield, created_at FROM plans WHERE id = $1"
+        "SELECT id, owner_address, token_address, amount, grace_period, grace_period_seconds, earn_yield, last_ping, is_active, status, yield_rate_bps, accrued_yield, created_at, onchain_plan_id FROM plans WHERE id = $1"
     )
     .bind(plan_id)
     .fetch_one(&state.db_pool)
@@ -960,6 +970,7 @@ async fn update_plan(
         yield_rate_bps: updated_plan_row.yield_rate_bps,
         accrued_yield: 0.0,
         created_at: updated_plan_row.created_at,
+        onchain_plan_id: updated_plan_row.onchain_plan_id,
         beneficiaries: inserted_beneficiaries,
     };
 
@@ -1012,7 +1023,7 @@ async fn get_plans(
                 r#"
                 SELECT id, owner_address, token_address, amount, grace_period,
                        grace_period_seconds, earn_yield, last_ping, is_active,
-                       status, yield_rate_bps, accrued_yield, created_at
+                       status, yield_rate_bps, accrued_yield, created_at, onchain_plan_id
                 FROM plans
                 WHERE owner_address = $1
                 ORDER BY created_at DESC
@@ -1040,7 +1051,7 @@ async fn get_plans(
                 r#"
                 SELECT DISTINCT p.id, p.owner_address, p.token_address, p.amount,
                        p.grace_period, p.grace_period_seconds, p.earn_yield,
-                       p.last_ping, p.is_active, p.status, p.yield_rate_bps, p.accrued_yield, p.created_at
+                       p.last_ping, p.is_active, p.status, p.yield_rate_bps, p.accrued_yield, p.created_at, p.onchain_plan_id
                 FROM plans p
                 INNER JOIN beneficiaries b ON b.plan_id = p.id
                 WHERE b.wallet_address = $1
@@ -1070,7 +1081,7 @@ async fn get_plans(
                 r#"
                 SELECT DISTINCT p.id, p.owner_address, p.token_address, p.amount,
                        p.grace_period, p.grace_period_seconds, p.earn_yield,
-                       p.last_ping, p.is_active, p.status, p.yield_rate_bps, p.accrued_yield, p.created_at
+                       p.last_ping, p.is_active, p.status, p.yield_rate_bps, p.accrued_yield, p.created_at, p.onchain_plan_id
                 FROM plans p
                 LEFT JOIN beneficiaries b ON b.plan_id = p.id
                 WHERE p.owner_address = $1 OR b.wallet_address = $2
@@ -1100,7 +1111,7 @@ async fn get_plans(
                 r#"
                 SELECT id, owner_address, token_address, amount, grace_period,
                        grace_period_seconds, earn_yield, last_ping, is_active,
-                       status, yield_rate_bps, accrued_yield, created_at
+                       status, yield_rate_bps, accrued_yield, created_at, onchain_plan_id
                 FROM plans
                 ORDER BY created_at DESC
                 "#,
@@ -1287,7 +1298,7 @@ async fn trigger_payout(
 
     // 2. Fetch the active plan for the owner
     let plan = match sqlx::query_as::<_, PlanRow>(
-        "SELECT id, owner_address, token_address, amount, grace_period, grace_period_seconds, earn_yield, last_ping, is_active, status, yield_rate_bps, accrued_yield, created_at FROM plans WHERE owner_address = $1 AND is_active = true FOR UPDATE",
+        "SELECT id, owner_address, token_address, amount, grace_period, grace_period_seconds, earn_yield, last_ping, is_active, status, yield_rate_bps, accrued_yield, created_at, onchain_plan_id FROM plans WHERE owner_address = $1 AND is_active = true FOR UPDATE",
     )
     .bind(&payload.owner)
     .fetch_optional(&mut *tx)
@@ -2020,7 +2031,7 @@ pub async fn get_plan_report(
         r#"
         SELECT id, owner_address, token_address, amount, grace_period,
                grace_period_seconds, earn_yield, last_ping, is_active,
-               status, yield_rate_bps, accrued_yield, created_at
+               status, yield_rate_bps, accrued_yield, created_at, onchain_plan_id
         FROM plans WHERE id = $1
         "#,
     )

backend/migrations/20260820000000_add_onchain_trigger_fields.up.sql

@@ -319,6 +319,7 @@ mod tests {
             yield_rate_bps: 500,
             accrued_yield: 42.5,
             created_at: Utc::now(),
+            onchain_plan_id: Some(7),
             beneficiaries: vec![BeneficiaryResponse {
                 id: Uuid::new_v4(),
                 plan_id: Uuid::new_v4(),