Validate history ttl to be less or equal to history meta ttl (#773)

Author: FZambia

internal/rule/rule.go

@@ -81,15 +81,21 @@ type Config struct {
 	// ClientConnectionRateLimit sets the maximum number of new connections a single Centrifugo
 	// node will accept per second.
 	ClientConnectionRateLimit int
+
+	// GlobalHistoryMetaTTL from here is used only for validation.
+	GlobalHistoryMetaTTL time.Duration
 }
 
+const DefaultGlobalHistoryMetaTTL = 30 * 24 * time.Hour
+
 // DefaultConfig has default config options.
 var DefaultConfig = Config{
 	ChannelPrivatePrefix:     "$", // so private channel will look like "$gossips".
 	ChannelNamespaceBoundary: ":", // so namespace "public" can be used as "public:news".
 	ChannelUserBoundary:      "#", // so user limited channel is "user#2694" where "2696" is user ID.
 	ChannelUserSeparator:     ",", // so several users limited channel is "dialog#2694,3019".
 	RpcNamespaceBoundary:     ":", // so rpc namespace "chat" can be used as "chat:get_user_info".
+	GlobalHistoryMetaTTL:     DefaultGlobalHistoryMetaTTL,
 }
 
 func stringInSlice(a string, list []string) bool {
@@ -104,13 +110,13 @@ func stringInSlice(a string, list []string) bool {
 var namePattern = "^[-a-zA-Z0-9_.]{2,}$"
 var nameRe = regexp.MustCompile(namePattern)
 
-func ValidateNamespace(ns ChannelNamespace) error {
+func ValidateNamespace(ns ChannelNamespace, globalHistoryMetaTTL time.Duration) error {
 	name := ns.Name
 	match := nameRe.MatchString(name)
 	if !match {
 		return fmt.Errorf("invalid namespace name – %s (must match %s regular expression)", name, namePattern)
 	}
-	if err := ValidateChannelOptions(ns.ChannelOptions); err != nil {
+	if err := ValidateChannelOptions(ns.ChannelOptions, globalHistoryMetaTTL); err != nil {
 		return err
 	}
 	return nil
@@ -128,10 +134,17 @@ func ValidateRpcNamespace(ns RpcNamespace) error {
 	return nil
 }
 
-func ValidateChannelOptions(c ChannelOptions) error {
+func ValidateChannelOptions(c ChannelOptions, globalHistoryMetaTTL time.Duration) error {
 	if (c.HistorySize != 0 && c.HistoryTTL == 0) || (c.HistorySize == 0 && c.HistoryTTL != 0) {
 		return errors.New("both history size and history ttl required for history")
 	}
+	historyMetaTTL := globalHistoryMetaTTL
+	if c.HistoryMetaTTL != 0 {
+		historyMetaTTL = time.Duration(c.HistoryMetaTTL)
+	}
+	if historyMetaTTL < time.Duration(c.HistoryTTL) {
+		return fmt.Errorf("history ttl (%s) can not be greater than history meta ttl (%s)", time.Duration(c.HistoryTTL), historyMetaTTL)
+	}
 	if c.ForceRecovery && (c.HistorySize == 0 || c.HistoryTTL == 0) {
 		return errors.New("both history size and history ttl required for recovery")
 	}
@@ -152,7 +165,7 @@ func ValidateRpcOptions(_ RpcOptions) error {
 
 // Validate validates config and returns error if problems found
 func (c *Config) Validate() error {
-	if err := ValidateChannelOptions(c.ChannelOptions); err != nil {
+	if err := ValidateChannelOptions(c.ChannelOptions, c.GlobalHistoryMetaTTL); err != nil {
 		return err
 	}
 	if err := ValidateRpcOptions(c.RpcOptions); err != nil {
@@ -175,7 +188,7 @@ func (c *Config) Validate() error {
 		if stringInSlice(n.Name, nss) {
 			return fmt.Errorf("namespace name must be unique: %s", n.Name)
 		}
-		if err := ValidateNamespace(n); err != nil {
+		if err := ValidateNamespace(n, c.GlobalHistoryMetaTTL); err != nil {
 			return fmt.Errorf("namespace %s: %v", n.Name, err)
 		}
 		if n.Name == personalChannelNamespace {

internal/rule/rule_test.go

@@ -6,6 +6,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/centrifugal/centrifugo/v5/internal/tools"
+
 	"github.com/stretchr/testify/require"
 )
 
@@ -100,6 +102,39 @@ func TestConfigValidatePersonalSingleConnectionOK(t *testing.T) {
 	require.NoError(t, err)
 }
 
+func TestConfigValidateHistoryTTL(t *testing.T) {
+	t.Run("in_namespace", func(t *testing.T) {
+		c := DefaultConfig
+		c.Namespaces = []ChannelNamespace{
+			{
+				Name: "name1",
+				ChannelOptions: ChannelOptions{
+					HistorySize:    10,
+					HistoryTTL:     tools.Duration(20 * time.Second),
+					HistoryMetaTTL: tools.Duration(10 * time.Second),
+				},
+			},
+		}
+		err := c.Validate()
+		require.ErrorContains(t, err, "history meta ttl")
+	})
+	t.Run("on_top_level", func(t *testing.T) {
+		c := DefaultConfig
+		c.HistorySize = 10
+		c.HistoryTTL = tools.Duration(31 * 24 * time.Hour)
+		err := c.Validate()
+		require.ErrorContains(t, err, "history meta ttl")
+	})
+	t.Run("top_level_non_default_global", func(t *testing.T) {
+		c := DefaultConfig
+		c.GlobalHistoryMetaTTL = 10 * time.Hour
+		c.HistorySize = 10
+		c.HistoryTTL = tools.Duration(30 * 24 * time.Hour)
+		err := c.Validate()
+		require.ErrorContains(t, err, "history meta ttl")
+	})
+}
+
 func TestConfigValidatePersonalSingleConnectionNamespacedFail(t *testing.T) {
 	c := DefaultConfig
 	c.Namespaces = []ChannelNamespace{}

main.go

@@ -134,7 +134,7 @@ var defaults = map[string]any{
 
 	"allowed_origins": []string{},
 
-	"global_history_meta_ttl":            30 * 24 * time.Hour,
+	"global_history_meta_ttl":            rule.DefaultGlobalHistoryMetaTTL,
 	"global_presence_ttl":                60 * time.Second,
 	"global_redis_presence_user_mapping": false,
 
@@ -1726,6 +1726,8 @@ func ruleConfig() rule.Config {
 	cfg.SubRefreshProxyName = v.GetString("sub_refresh_proxy_name")
 	cfg.ProxySubscribeStream = v.GetBool("proxy_stream_subscribe")
 	cfg.ProxySubscribeStreamBidirectional = v.GetBool("proxy_subscribe_stream_bidirectional")
+	// GlobalHistoryMetaTTL is required here only for validation purposes.
+	cfg.GlobalHistoryMetaTTL = GetDuration("global_history_meta_ttl", true)
 
 	cfg.Namespaces = namespacesFromConfig(v)