config: add event subscription options and defaults (#7930)

CHANGELOG_PENDING.md

@@ -17,6 +17,7 @@ Special thanks to external contributors on this release:
   - [mempool] \#7171 Remove legacy mempool implementation. (@tychoish)
   - [rpc] \#7575 Rework how RPC responses are written back via HTTP. (@creachadair)
   - [rpc] \#7713 Remove unused options for websocket clients. (@creachadair)
+  - [config] \#7930 Add new event subscription options and defaults. (@creachadair)
 
 - Apps
 

config/config.go

@@ -442,6 +442,33 @@ type RPCConfig struct {
 	// to the estimated maximum number of broadcast_tx_commit calls per block.
 	MaxSubscriptionsPerClient int `mapstructure:"max-subscriptions-per-client"`
 
+	// If true, disable the websocket interface to the RPC service.  This has
+	// the effect of disabling the /subscribe, /unsubscribe, and /unsubscribe_all
+	// methods for event subscription.
+	//
+	// EXPERIMENTAL: This setting will be removed in Tendermint v0.37.
+	ExperimentalDisableWebsocket bool `mapstructure:"experimental-disable-websocket"`
+
+	// The time window size for the event log. All events up to this long before
+	// the latest (up to EventLogMaxItems) will be available for subscribers to
+	// fetch via the /events method.  If 0 (the default) the event log and the
+	// /events RPC method are disabled.
+	EventLogWindowSize time.Duration `mapstructure:"event-log-window-size"`
+
+	// The maxiumum number of events that may be retained by the event log.  If
+	// this value is 0, no upper limit is set. Otherwise, items in excess of
+	// this number will be discarded from the event log.
+	//
+	// Warning: This setting is a safety valve. Setting it too low may cause
+	// subscribers to miss events.  Try to choose a value higher than the
+	// maximum worst-case expected event load within the chosen window size in
+	// ordinary operation.
+	//
+	// For example, if the window size is 10 minutes and the node typically
+	// averages 1000 events per ten minutes, but with occasional known spikes of
+	// up to 2000, choose a value > 2000.
+	EventLogMaxItems int `mapstructure:"event-log-max-items"`
+
 	// How long to wait for a tx to be committed during /broadcast_tx_commit
 	// WARNING: Using a value larger than 10s will result in increasing the
 	// global HTTP write timeout, which applies to all connections and endpoints.
@@ -487,9 +514,14 @@ func DefaultRPCConfig() *RPCConfig {
 		Unsafe:             false,
 		MaxOpenConnections: 900,
 
-		MaxSubscriptionClients:    100,
-		MaxSubscriptionsPerClient: 5,
-		TimeoutBroadcastTxCommit:  10 * time.Second,
+		// Settings for event subscription.
+		MaxSubscriptionClients:       100,
+		MaxSubscriptionsPerClient:    5,
+		ExperimentalDisableWebsocket: false, // compatible with TM v0.35 and earlier
+		EventLogWindowSize:           0,     // disables /events RPC by default
+		EventLogMaxItems:             0,
+
+		TimeoutBroadcastTxCommit: 10 * time.Second,
 
 		MaxBodyBytes:   int64(1000000), // 1MB
 		MaxHeaderBytes: 1 << 20,        // same as the net/http default
@@ -519,6 +551,12 @@ func (cfg *RPCConfig) ValidateBasic() error {
 	if cfg.MaxSubscriptionsPerClient < 0 {
 		return errors.New("max-subscriptions-per-client can't be negative")
 	}
+	if cfg.EventLogWindowSize < 0 {
+		return errors.New("event-log-window-size must not be negative")
+	}
+	if cfg.EventLogMaxItems < 0 {
+		return errors.New("event-log-max-items must not be negative")
+	}
 	if cfg.TimeoutBroadcastTxCommit < 0 {
 		return errors.New("timeout-broadcast-tx-commit can't be negative")
 	}

config/toml.go

@@ -220,6 +220,33 @@ max-subscription-clients = {{ .RPC.MaxSubscriptionClients }}
 # to the estimated maximum number of broadcast_tx_commit calls per block.
 max-subscriptions-per-client = {{ .RPC.MaxSubscriptionsPerClient }}
 
+# If true, disable the websocket interface to the RPC service.  This has
+# the effect of disabling the /subscribe, /unsubscribe, and /unsubscribe_all
+# methods for event subscription.
+#
+# EXPERIMENTAL: This setting will be removed in Tendermint v0.37.
+experimental-disable-websocket = {{ .RPC.ExperimentalDisableWebsocket }}
+
+# The time window size for the event log. All events up to this long before
+# the latest (up to EventLogMaxItems) will be available for subscribers to
+# fetch via the /events method.  If 0 (the default) the event log and the
+# /events RPC method are disabled.
+event-log-window-size = "{{ .RPC.EventLogWindowSize }}"
+
+# The maxiumum number of events that may be retained by the event log.  If
+# this value is 0, no upper limit is set. Otherwise, items in excess of
+# this number will be discarded from the event log.
+#
+# Warning: This setting is a safety valve. Setting it too low may cause
+# subscribers to miss events.  Try to choose a value higher than the
+# maximum worst-case expected event load within the chosen window size in
+# ordinary operation.
+#
+# For example, if the window size is 10 minutes and the node typically
+# averages 1000 events per ten minutes, but with occasional known spikes of
+# up to 2000, choose a value > 2000.
+event-log-max-items = {{ .RPC.EventLogMaxItems }}
+
 # How long to wait for a tx to be committed during /broadcast_tx_commit.
 # WARNING: Using a value larger than 10s will result in increasing the
 # global HTTP write timeout, which applies to all connections and endpoints.

internal/rpc/core/env.go

@@ -239,23 +239,29 @@ func (env *Environment) StartService(ctx context.Context, conf *config.Config) (
 		cfg.WriteTimeout = conf.RPC.TimeoutBroadcastTxCommit + 1*time.Second
 	}
 
-	// we may expose the rpc over both a unix and tcp socket
+	// We may expose the RPC over both TCP and a Unix-domain socket.
 	listeners := make([]net.Listener, len(listenAddrs))
 	for i, listenAddr := range listenAddrs {
 		mux := http.NewServeMux()
 		rpcLogger := env.Logger.With("module", "rpc-server")
-		wmLogger := rpcLogger.With("protocol", "websocket")
-		wm := rpcserver.NewWebsocketManager(wmLogger, routes,
-			rpcserver.OnDisconnect(func(remoteAddr string) {
-				err := env.EventBus.UnsubscribeAll(context.Background(), remoteAddr)
-				if err != nil && err != tmpubsub.ErrSubscriptionNotFound {
-					wmLogger.Error("Failed to unsubscribe addr from events", "addr", remoteAddr, "err", err)
-				}
-			}),
-			rpcserver.ReadLimit(cfg.MaxBodyBytes),
-		)
-		mux.HandleFunc("/websocket", wm.WebsocketHandler)
 		rpcserver.RegisterRPCFuncs(mux, routes, rpcLogger)
+
+		if conf.RPC.ExperimentalDisableWebsocket {
+			rpcLogger.Info("Disabling websocket endpoints (experimental-disable-websocket=true)")
+		} else {
+			wmLogger := rpcLogger.With("protocol", "websocket")
+			wm := rpcserver.NewWebsocketManager(wmLogger, routes,
+				rpcserver.OnDisconnect(func(remoteAddr string) {
+					err := env.EventBus.UnsubscribeAll(context.Background(), remoteAddr)
+					if err != nil && err != tmpubsub.ErrSubscriptionNotFound {
+						wmLogger.Error("Failed to unsubscribe addr from events", "addr", remoteAddr, "err", err)
+					}
+				}),
+				rpcserver.ReadLimit(cfg.MaxBodyBytes),
+			)
+			mux.HandleFunc("/websocket", wm.WebsocketHandler)
+		}
+
 		listener, err := rpcserver.Listen(
 			listenAddr,
 			cfg.MaxOpenConnections,