feat: add CEL-based batch_trigger gate for round start

README.md

@@ -109,6 +109,7 @@ The `arkd` server can be configured using environment variables.
 | `ARKD_INDEXER_EXPOSURE`.           | Require intent for getting vtxo chain (public, private, withheld)               | `public`                       |
 | `ARKD_INDEXER_SIGNING_PRIVKEY`     | Hex-encoded private key for indexer auth token signing (sensitive)              | -                              |
 | `ARKD_INDEXER_AUTH_TOKEN_EXPIRY`   | Auth token TTL in seconds                                                       | `300` (5 minutes)              |
+| `ARKD_BATCH_TRIGGER`               | Optional CEL formula returning `bool`. When set, the server only starts a new batch round when the formula evaluates to `true`. See [`pkg/ark-lib/batchtrigger/README.md`](pkg/ark-lib/batchtrigger/README.md) for the available variables and examples. | - (always start)               |
 
 ## Provisioning
 

envs/arkd.dev.env

@@ -17,3 +17,7 @@ ARKD_BAN_THRESHOLD=1
 ARKD_ONCHAIN_OUTPUT_FEE=100
 ARKD_ENABLE_PPROF=true
 ARKD_UNROLLED_VTXO_MIN_EXPIRY_MARGIN=10
+# Optional CEL gate: only start a batch round when the formula returns true.
+# Empty/unset = always start (legacy behaviour).
+# See pkg/ark-lib/batchtrigger/README.md for the variable reference.
+ARKD_BATCH_TRIGGER=

envs/arkd.light.env

@@ -16,3 +16,7 @@ ARKD_ONCHAIN_OUTPUT_FEE=100
 ARKD_VTXO_TREE_EXPIRY=40
 ARKD_CHECKPOINT_EXIT_DELAY=10
 ARKD_UNROLLED_VTXO_MIN_EXPIRY_MARGIN=10
+# Optional CEL gate: only start a batch round when the formula returns true.
+# Empty/unset = always start (legacy behaviour).
+# See pkg/ark-lib/batchtrigger/README.md for the variable reference.
+ARKD_BATCH_TRIGGER=

internal/config/config.go

@@ -26,6 +26,7 @@ import (
 	fileunlocker "github.com/arkade-os/arkd/internal/infrastructure/unlocker/file"
 	walletclient "github.com/arkade-os/arkd/internal/infrastructure/wallet"
 	arklib "github.com/arkade-os/arkd/pkg/ark-lib"
+	"github.com/arkade-os/arkd/pkg/ark-lib/batchtrigger"
 	"github.com/btcsuite/btcd/btcec/v2"
 	"github.com/redis/go-redis/v9"
 	log "github.com/sirupsen/logrus"
@@ -143,6 +144,11 @@ type Config struct {
 	IndexerSigningKey    string
 	MaxConcurrentStreams uint32
 
+	// BatchTrigger is an optional CEL formula. When set, the server only
+	// starts a new batch round when the formula evaluates to true. When
+	// empty, every session starts a round (legacy behaviour).
+	BatchTrigger string
+
 	fee            ports.FeeManager
 	repo           ports.RepoManager
 	svc            application.Service
@@ -243,6 +249,9 @@ var (
 	// IndexerSigningKey is a hex-encoded private key. SENSITIVE: never log this value.
 	IndexerSigningKey    = "INDEXER_SIGNING_PRIVKEY" // #nosec G101
 	MaxConcurrentStreams = "MAX_CONCURRENT_STREAMS"
+	// BatchTrigger is a CEL formula evaluated before every round to decide
+	// whether the server should start a new batch. Empty = always start.
+	BatchTrigger = "BATCH_TRIGGER"
 
 	defaultDatadir             = arklib.AppDataDir("arkd", false)
 	defaultSessionDuration     = 30
@@ -443,6 +452,7 @@ func LoadConfig() (*Config, error) {
 		MaxConcurrentStreams:          viper.GetUint32(MaxConcurrentStreams),
 		// Default to 1 if set to 0
 		MaxOpReturnOutputs: max(1, viper.GetUint32(MaxOpReturnOutputs)),
+		BatchTrigger:       viper.GetString(BatchTrigger),
 	}, nil
 }
 
@@ -651,6 +661,10 @@ func (c *Config) Validate() error {
 		return fmt.Errorf("max concurrent streams must be greater than 0")
 	}
 
+	if _, err := batchtrigger.New(c.BatchTrigger); err != nil {
+		return fmt.Errorf("invalid batch trigger program: %w", err)
+	}
+
 	if err := c.repoManager(); err != nil {
 		return err
 	}
@@ -977,6 +991,7 @@ func (c *Config) appService() error {
 		c.SettlementMinExpiryGap,
 		c.UnrolledVtxoMinExpiryMargin,
 		time.Unix(c.VtxoNoCsvValidationCutoffDate, 0), c.MaxOpReturnOutputs,
+		c.BatchTrigger,
 	)
 	if err != nil {
 		return err

internal/core/application/batch_trigger_test.go

@@ -0,0 +1,222 @@
+package application
+
+import (
+	"testing"
+	"time"
+
+	"github.com/arkade-os/arkd/internal/core/domain"
+	"github.com/arkade-os/arkd/internal/core/ports"
+	"github.com/arkade-os/arkd/pkg/ark-lib/batchtrigger"
+	"github.com/stretchr/testify/require"
+)
+
+func TestAggregateIntentTriggerData(t *testing.T) {
+	tests := []struct {
+		name                    string
+		intents                 []ports.TimedIntent
+		wantBoardingInputsCount int64
+		wantTotalBoardingAmount uint64
+		wantTotalIntentFees     uint64
+	}{
+		{
+			name:                    "empty intents",
+			intents:                 nil,
+			wantBoardingInputsCount: 0,
+			wantTotalBoardingAmount: 0,
+			wantTotalIntentFees:     0,
+		},
+		{
+			name: "single intent with boarding inputs and positive fee",
+			intents: []ports.TimedIntent{
+				{
+					Intent: domain.Intent{
+						Inputs: []domain.Vtxo{
+							{Amount: 1000},
+							{Amount: 500},
+						},
+						Receivers: []domain.Receiver{
+							{Amount: 800},
+							{Amount: 600},
+						},
+					},
+					BoardingInputs: []ports.BoardingInput{
+						{Amount: 200},
+						{Amount: 300},
+					},
+				},
+			},
+			wantBoardingInputsCount: 2,
+			wantTotalBoardingAmount: 500,
+			// inputs: 1500 vtxo + 500 boarding = 2000; outputs: 1400; fee = 600
+			wantTotalIntentFees: 600,
+		},
+		{
+			name: "intent with no boarding and no fee (inputs == outputs)",
+			intents: []ports.TimedIntent{
+				{
+					Intent: domain.Intent{
+						Inputs:    []domain.Vtxo{{Amount: 1000}},
+						Receivers: []domain.Receiver{{Amount: 1000}},
+					},
+				},
+			},
+			wantBoardingInputsCount: 0,
+			wantTotalBoardingAmount: 0,
+			wantTotalIntentFees:     0,
+		},
+		{
+			name: "intent where outputs exceed inputs is treated as zero fee",
+			intents: []ports.TimedIntent{
+				{
+					Intent: domain.Intent{
+						Inputs:    []domain.Vtxo{{Amount: 100}},
+						Receivers: []domain.Receiver{{Amount: 200}},
+					},
+				},
+			},
+			wantBoardingInputsCount: 0,
+			wantTotalBoardingAmount: 0,
+			wantTotalIntentFees:     0,
+		},
+		{
+			name: "multiple intents are summed",
+			intents: []ports.TimedIntent{
+				{
+					Intent: domain.Intent{
+						Inputs:    []domain.Vtxo{{Amount: 1000}},
+						Receivers: []domain.Receiver{{Amount: 900}},
+					},
+					BoardingInputs: []ports.BoardingInput{{Amount: 50}},
+				},
+				{
+					Intent: domain.Intent{
+						Inputs:    []domain.Vtxo{{Amount: 2000}},
+						Receivers: []domain.Receiver{{Amount: 1800}},
+					},
+					BoardingInputs: []ports.BoardingInput{
+						{Amount: 100},
+						{Amount: 100},
+					},
+				},
+			},
+			wantBoardingInputsCount: 3,
+			wantTotalBoardingAmount: 250,
+			// intent 1: 1000+50 - 900 = 150
+			// intent 2: 2000+200 - 1800 = 400
+			wantTotalIntentFees: 550,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			gotCount, gotBoarding, gotFees := aggregateIntentTriggerData(tt.intents)
+			require.Equal(t, tt.wantBoardingInputsCount, gotCount, "boarding inputs count")
+			require.Equal(t, tt.wantTotalBoardingAmount, gotBoarding, "total boarding amount")
+			require.Equal(t, tt.wantTotalIntentFees, gotFees, "total intent fees")
+		})
+	}
+}
+
+func TestEvalBatchTrigger(t *testing.T) {
+	tests := []struct {
+		name    string
+		program string
+		ctx     batchtrigger.Context
+		want    bool
+	}{
+		{
+			name:    "nil trigger always permits",
+			program: "",
+			want:    true,
+		},
+		{
+			name:    "true literal permits",
+			program: "true",
+			want:    true,
+		},
+		{
+			name:    "false literal denies",
+			program: "false",
+			want:    false,
+		},
+		{
+			name:    "intent count gate satisfied",
+			program: "intents_count >= 2.0",
+			ctx:     batchtrigger.Context{IntentsCount: 5},
+			want:    true,
+		},
+		{
+			name:    "intent count gate unsatisfied",
+			program: "intents_count >= 2.0",
+			ctx:     batchtrigger.Context{IntentsCount: 1},
+			want:    false,
+		},
+		{
+			name:    "fee revenue gate satisfied",
+			program: "total_intent_fees >= 500.0",
+			ctx:     batchtrigger.Context{TotalIntentFees: 1000},
+			want:    true,
+		},
+		{
+			name: "issue 1045 example: many intents, low fees",
+			program: "intents_count > 1.0 && " +
+				"(current_feerate <= 2.0 || time_since_last_batch >= 3600.0)",
+			ctx: batchtrigger.Context{
+				IntentsCount:   3,
+				CurrentFeerate: 1,
+			},
+			want: true,
+		},
+		{
+			name: "issue 1045 example: many intents, high fees, but stale",
+			program: "intents_count > 1.0 && " +
+				"(current_feerate <= 2.0 || time_since_last_batch >= 3600.0)",
+			ctx: batchtrigger.Context{
+				IntentsCount:       3,
+				CurrentFeerate:     50,
+				TimeSinceLastBatch: 7200,
+			},
+			want: true,
+		},
+		{
+			name: "issue 1045 example: many intents, high fees, recent",
+			program: "intents_count > 1.0 && " +
+				"(current_feerate <= 2.0 || time_since_last_batch >= 3600.0)",
+			ctx: batchtrigger.Context{
+				IntentsCount:       3,
+				CurrentFeerate:     50,
+				TimeSinceLastBatch: 60,
+			},
+			want: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tr, err := batchtrigger.New(tt.program)
+			require.NoError(t, err)
+
+			s := &service{batchTrigger: tr}
+			require.Equal(t, tt.want, s.evalBatchTrigger(tt.ctx))
+		})
+	}
+}
+
+func TestEvalBatchTriggerNilFailsOpen(t *testing.T) {
+	// A nil service.batchTrigger must permit; the context value is ignored.
+	s := &service{}
+	require.True(t, s.evalBatchTrigger(batchtrigger.Context{}))
+	require.True(t, s.evalBatchTrigger(batchtrigger.Context{IntentsCount: 0}))
+}
+
+func TestLastBatchAtRoundtrip(t *testing.T) {
+	// Sanity check that the atomic counter we use to derive
+	// time_since_last_batch behaves as expected: zero until set, monotonic
+	// once written.
+	s := &service{}
+	require.Equal(t, int64(0), s.lastBatchAt.Load())
+
+	now := time.Now().Unix()
+	s.lastBatchAt.Store(now)
+	require.Equal(t, now, s.lastBatchAt.Load())
+}