Implement optional StreamMaxLifetime for server streams

README.md

@@ -101,6 +101,7 @@ The `arkd` server can be configured using environment variables and the admin se
 | `ARKD_OTEL_COLLECTOR_ENDPOINT`     | OpenTelemetry collector endpoint                                                | -                              |
 | `ARKD_OTEL_PUSH_INTERVAL`          | OpenTelemetry push interval in seconds                                          | `10`                           |
 | `ARKD_HEARTBEAT_INTERVAL`          | Heartbeat interval in seconds                                                   | `60`                           |
+| `ARKD_STREAM_MAX_LIFETIME`         | Max lifetime of a server-streaming RPC in seconds (`0` disables)                | `1800`                         |
 | `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)              |

cmd/arkd/main.go

@@ -43,6 +43,7 @@ func startAction(_ *cli.Context) error {
 		TLSExtraIPs:          cfg.TLSExtraIPs,
 		TLSExtraDomains:      cfg.TLSExtraDomains,
 		HeartbeatInterval:    cfg.HeartbeatInterval,
+		StreamMaxLifetime:    cfg.StreamMaxLifetime,
 		EnablePprof:          cfg.EnablePprof,
 		EnableChannelz:       cfg.EnableChannelz,
 		MaxConcurrentStreams: cfg.MaxConcurrentStreams,

internal/config/config.go

@@ -103,6 +103,7 @@ type Config struct {
 	BoardingExitDelay          arklib.RelativeLocktime
 	NoteUriPrefix              string
 	HeartbeatInterval          int64
+	StreamMaxLifetime          int64
 	BuildVersionHeaderRequired bool
 	BuildVersionHeader         string
 	DigestHeaderRequired       bool
@@ -230,6 +231,7 @@ var (
 	UtxoMinAmount             = "UTXO_MIN_AMOUNT"
 	VtxoMinAmount             = "VTXO_MIN_AMOUNT"
 	HeartbeatInterval         = "HEARTBEAT_INTERVAL"
+	StreamMaxLifetime         = "STREAM_MAX_LIFETIME"
 	SettlementMinExpiryGap    = "SETTLEMENT_MIN_EXPIRY_GAP"
 	// Minimum remaining CSV time (in seconds) for an unrolled VTXO to be accepted into a batch.
 	// 0 means fallback to session duration.
@@ -290,10 +292,11 @@ var (
 
 	defaultRoundMaxParticipantsCount     = 128
 	defaultRoundMinParticipantsCount     = 1
-	defaultOtelPushInterval              = 10  // seconds
-	defaultHeartbeatInterval             = 60  // seconds
-	defaultSettlementMinExpiryGap        = 0   // disabled by default
-	defaultUnrolledVtxoMinExpiryMargin   = 300 // 5 minutes in seconds
+	defaultOtelPushInterval              = 10   // seconds
+	defaultHeartbeatInterval             = 60   // seconds
+	defaultStreamMaxLifetime             = 1800 // 30 minutes in seconds
+	defaultSettlementMinExpiryGap        = 0    // disabled by default
+	defaultUnrolledVtxoMinExpiryMargin   = 300  // 5 minutes in seconds
 	defaultMaxTxWeight                   = int64(0.01 * bitcoinBlockWeight)
 	defaultAssetTxMaxWeightRatio         = 0.5
 	defaultVtxoNoCsvValidationCutoffDate = 0 // disabled by default
@@ -346,6 +349,7 @@ func LoadConfig() (*Config, error) {
 	viper.SetDefault(RedisTxNumOfRetries, defaultRedisTxNumOfRetries)
 	viper.SetDefault(OtelPushInterval, defaultOtelPushInterval)
 	viper.SetDefault(HeartbeatInterval, defaultHeartbeatInterval)
+	viper.SetDefault(StreamMaxLifetime, defaultStreamMaxLifetime)
 	viper.SetDefault(SettlementMinExpiryGap, defaultSettlementMinExpiryGap)
 	viper.SetDefault(UnrolledVtxoMinExpiryMargin, defaultUnrolledVtxoMinExpiryMargin)
 	viper.SetDefault(MaxTxWeight, defaultMaxTxWeight)
@@ -491,6 +495,7 @@ func LoadConfig() (*Config, error) {
 		OtelPushInterval:          viper.GetInt64(OtelPushInterval),
 		PyroscopeServerURL:        viper.GetString(PyroscopeServerURL),
 		HeartbeatInterval:         viper.GetInt64(HeartbeatInterval),
+		StreamMaxLifetime:         viper.GetInt64(StreamMaxLifetime),
 
 		RoundMaxParticipantsCount:     viper.GetUint64(RoundMaxParticipantsCount),
 		RoundMinParticipantsCount:     viper.GetUint64(RoundMinParticipantsCount),

internal/interface/grpc/config.go

@@ -19,6 +19,7 @@ type Config struct {
 	TLSExtraIPs          []string
 	TLSExtraDomains      []string
 	HeartbeatInterval    int64
+	StreamMaxLifetime    int64
 	EnablePprof          bool
 	EnableChannelz       bool
 	MaxConcurrentStreams uint32

internal/interface/grpc/handlers/arkservice.go

@@ -24,17 +24,25 @@ type service interface {
 type handler struct {
 	version   string
 	heartbeat time.Duration
+	// maxStreamLifetime bounds how long a single server-streaming RPC
+	// (GetEventStream / GetTransactionsStream) may stay open. Abandoned
+	// streams whose client disconnect is never observed are reaped after this
+	// duration; live clients reconnect transparently. Zero disables the bound.
+	maxStreamLifetime time.Duration
 
 	svc application.Service
 
 	eventsListenerHandler       *broker[*arkv1.GetEventStreamResponse]
 	transactionsListenerHandler *broker[*arkv1.GetTransactionsStreamResponse]
 }
 
-func NewAppServiceHandler(version string, service application.Service, heartbeat int64) service {
+func NewAppServiceHandler(
+	version string, service application.Service, heartbeat int64, maxStreamLifetime int64,
+) service {
 	h := &handler{
 		version:                     version,
 		heartbeat:                   time.Duration(heartbeat) * time.Second,
+		maxStreamLifetime:           time.Duration(maxStreamLifetime) * time.Second,
 		svc:                         service,
 		eventsListenerHandler:       newBroker[*arkv1.GetEventStreamResponse](),
 		transactionsListenerHandler: newBroker[*arkv1.GetTransactionsStreamResponse](),
@@ -226,6 +234,19 @@ func (h *handler) SubmitSignedForfeitTxs(
 	return &arkv1.SubmitSignedForfeitTxsResponse{}, nil
 }
 
+// streamContext derives a child of the stream's context bounded by an absolute
+// max lifetime. When maxLifetime <= 0 the bound is disabled and only the
+// stream's own cancellation applies. The returned cancel func must always be
+// called to release resources.
+func streamContext(
+	parent context.Context, maxLifetime time.Duration,
+) (context.Context, context.CancelFunc) {
+	if maxLifetime <= 0 {
+		return context.WithCancel(parent)
+	}
+	return context.WithTimeout(parent, maxLifetime)
+}
+
 func (h *handler) GetEventStream(
 	req *arkv1.GetEventStreamRequest, stream arkv1.ArkService_GetEventStreamServer,
 ) error {
@@ -247,6 +268,12 @@ func (h *handler) GetEventStream(
 		return err
 	}
 
+	// Bound the stream lifetime so abandoned subscriptions are reaped even if
+	// the client disconnect is never observed (e.g. masked by an upstream
+	// proxy). On expiry we return nil: the client sees io.EOF and reconnects.
+	ctx, cancel := streamContext(stream.Context(), h.maxStreamLifetime)
+	defer cancel()
+
 	// create a Timer that will fire after one heartbeat interval
 	timer := time.NewTimer(h.heartbeat)
 	defer timer.Stop()
@@ -265,7 +292,7 @@ func (h *handler) GetEventStream(
 
 	for {
 		select {
-		case <-stream.Context().Done():
+		case <-ctx.Done():
 			return nil
 		case <-listener.done:
 			return nil
@@ -437,6 +464,12 @@ func (h *handler) GetTransactionsStream(
 		h.transactionsListenerHandler.removeListener(listener.id)
 	}()
 
+	// Bound the stream lifetime so abandoned subscriptions are reaped even if
+	// the client disconnect is never observed (e.g. masked by an upstream
+	// proxy). On expiry we return nil: the client sees io.EOF and reconnects.
+	ctx, cancel := streamContext(stream.Context(), h.maxStreamLifetime)
+	defer cancel()
+
 	// create a Timer that will fire after one heartbeat interval
 	timer := time.NewTimer(h.heartbeat)
 	defer timer.Stop()
@@ -455,7 +488,7 @@ func (h *handler) GetTransactionsStream(
 
 	for {
 		select {
-		case <-stream.Context().Done():
+		case <-ctx.Done():
 			return nil
 		case <-listener.done:
 			return nil

internal/interface/grpc/handlers/indexer.go

@@ -33,18 +33,24 @@ type indexerService struct {
 	subscriptionTimeoutDuration time.Duration
 
 	heartbeat time.Duration
+	// maxStreamLifetime bounds how long a single GetSubscription stream may
+	// stay open. Abandoned subscriptions (client vanished behind a proxy that
+	// masks the disconnect) are reaped after this duration; live clients
+	// reconnect transparently. Zero disables the bound.
+	maxStreamLifetime time.Duration
 }
 
 func NewIndexerService(
 	indexerSvc application.IndexerService, eventsCh <-chan application.TransactionEvent,
-	subscriptionTimeoutDuration time.Duration, heartbeat int64,
+	subscriptionTimeoutDuration time.Duration, heartbeat int64, maxStreamLifetime int64,
 ) arkv1.IndexerServiceServer {
 	svc := &indexerService{
 		indexerSvc:                  indexerSvc,
 		eventsCh:                    eventsCh,
 		scriptSubsHandler:           newBroker[*arkv1.GetSubscriptionResponse](),
 		subscriptionTimeoutDuration: subscriptionTimeoutDuration,
 		heartbeat:                   time.Duration(heartbeat) * time.Second,
+		maxStreamLifetime:           time.Duration(maxStreamLifetime) * time.Second,
 	}
 
 	go svc.listenToTxEvents()
@@ -491,6 +497,12 @@ func (h *indexerService) GetSubscription(
 		}
 	}
 
+	// Bound the stream lifetime so abandoned subscriptions are reaped even if
+	// the client disconnect is never observed (e.g. masked by an upstream
+	// proxy). On expiry we return nil: the client sees io.EOF and reconnects.
+	ctx, cancel := streamContext(stream.Context(), h.maxStreamLifetime)
+	defer cancel()
+
 	// create a Timer that will fire after one heartbeat interval
 	timer := time.NewTimer(h.heartbeat)
 	defer timer.Stop()
@@ -517,7 +529,7 @@ func (h *indexerService) GetSubscription(
 		// pending, it returns immediately without draining listener.ch,
 		// leaving buffered events for the successor.
 		select {
-		case <-stream.Context().Done():
+		case <-ctx.Done():
 			return nil
 		case <-listener.done:
 			return nil

internal/interface/grpc/handlers/indexer_test.go

@@ -1832,6 +1832,76 @@ func TestSubscribeForScripts(t *testing.T) {
 	})
 }
 
+// TestGetSubscriptionMaxLifetime asserts that a streaming subscription is
+// proactively closed once maxStreamLifetime elapses, even when the client
+// never disconnects (its context stays live). This reaps abandoned streams
+// that would otherwise pile up on the gateway loopback sockets; the client
+// SDK reconnects transparently on the resulting io.EOF.
+func TestGetSubscriptionMaxLifetime(t *testing.T) {
+	svc := newTestIndexerService(t)
+	svc.maxStreamLifetime = 150 * time.Millisecond
+
+	// A context that never cancels on its own: only the max-lifetime bound
+	// should end the stream.
+	stream := newMockGetSubscriptionServer(context.Background())
+
+	errCh := make(chan error, 1)
+	start := time.Now()
+	go func() {
+		errCh <- svc.GetSubscription(&arkv1.GetSubscriptionRequest{}, stream)
+	}()
+
+	// Drain the initial SubscriptionStarted event.
+	require.NotNil(t, stream.recv(t, time.Second).GetSubscriptionStarted())
+
+	select {
+	case err := <-errCh:
+		require.NoError(t, err, "handler must close gracefully (nil) so the client reconnects")
+		require.GreaterOrEqual(t, time.Since(start), 150*time.Millisecond,
+			"stream must live at least its max lifetime")
+	case <-time.After(2 * time.Second):
+		t.Fatal("GetSubscription did not return after max stream lifetime elapsed")
+	}
+}
+
+// TestStreamContext locks the contract of the streamContext helper shared by
+// all three streaming handlers: zero disables the bound (cancel tracks the
+// parent only), a positive value imposes an absolute deadline.
+func TestStreamContext(t *testing.T) {
+	t.Run("disabled when zero", func(t *testing.T) {
+		parent, cancelParent := context.WithCancel(context.Background())
+		defer cancelParent()
+
+		ctx, cancel := streamContext(parent, 0)
+		defer cancel()
+
+		_, hasDeadline := ctx.Deadline()
+		require.False(t, hasDeadline, "no bound expected when maxLifetime is zero")
+
+		cancelParent()
+		select {
+		case <-ctx.Done():
+		case <-time.After(time.Second):
+			t.Fatal("ctx must cancel together with its parent")
+		}
+	})
+
+	t.Run("bounded when positive", func(t *testing.T) {
+		ctx, cancel := streamContext(context.Background(), 100*time.Millisecond)
+		defer cancel()
+
+		_, hasDeadline := ctx.Deadline()
+		require.True(t, hasDeadline, "deadline expected when maxLifetime is positive")
+
+		select {
+		case <-ctx.Done():
+			require.ErrorIs(t, ctx.Err(), context.DeadlineExceeded)
+		case <-time.After(time.Second):
+			t.Fatal("ctx must expire after its max lifetime")
+		}
+	})
+}
+
 // requireSubscriptionNotFound asserts err is the structured
 // SUBSCRIPTION_NOT_FOUND error: it exposes the NotFound gRPC code and the
 // SUBSCRIPTION_NOT_FOUND structured code, and preserves the legacy

internal/interface/grpc/service.go

@@ -345,7 +345,9 @@ func (s *service) newServer(tlsConfig *tls.Config, withPprof, withChannelz bool)
 	if err != nil {
 		return fmt.Errorf("failed to create app service: %w", err)
 	}
-	appHandler := handlers.NewAppServiceHandler(s.version, appSvc, s.config.HeartbeatInterval)
+	appHandler := handlers.NewAppServiceHandler(
+		s.version, appSvc, s.config.HeartbeatInterval, s.config.StreamMaxLifetime,
+	)
 	eventsCh := appSvc.GetIndexerTxChannel(ctx)
 	subscriptionTimeoutDuration := time.Minute
 	indexerSvc, err := s.appConfig.IndexerService()
@@ -357,6 +359,7 @@ func (s *service) newServer(tlsConfig *tls.Config, withPprof, withChannelz bool)
 		eventsCh,
 		subscriptionTimeoutDuration,
 		s.config.HeartbeatInterval,
+		s.config.StreamMaxLifetime,
 	)
 	arkv1.RegisterArkServiceServer(grpcServer, appHandler)
 	arkv1.RegisterIndexerServiceServer(grpcServer, indexerHandler)