Merge pull request #1504 from c3llus/feat/issue-1499/new-max-defer-delay

nsqd: add support for configurable max-defer-timeout for DPUB

Author: mreiferson

apps/nsqd/main.go

@@ -54,6 +54,7 @@ func (p *program) Init(env svc.Environment) error {
 	cfg.Validate()
 
 	options.Resolve(opts, flagSet, cfg)
+	applyBackwardCompatibility(opts, flagSet)
 
 	nsqd, err := nsqd.New(opts)
 	if err != nil {
@@ -104,3 +105,13 @@ func (p *program) Context() context.Context {
 func logFatal(f string, args ...interface{}) {
 	lg.LogFatal("[nsqd] ", f, args...)
 }
+
+// applyBackwardCompatibility applies backward compatibility rules to options after flag resolution
+func applyBackwardCompatibility(opts *nsqd.Options, flagSet *flag.FlagSet) {
+	// when max-defer-timeout was not explicitly set, refer to the max-req-timeout value
+	if flag := flagSet.Lookup("max-defer-timeout"); flag != nil && flag.Value.String() == flag.DefValue {
+		opts.MaxDeferTimeout = opts.MaxReqTimeout
+	}
+
+	// ... other backward compatibility rules can be added here
+}

apps/nsqd/options.go

@@ -161,6 +161,7 @@ func nsqdFlagSet(opts *nsqd.Options) *flag.FlagSet {
 	flagSet.Int64("max-msg-size", opts.MaxMsgSize, "maximum size of a single message in bytes")
 	flagSet.Duration("max-req-timeout", opts.MaxReqTimeout, "maximum requeuing timeout for a message")
 	flagSet.Int64("max-body-size", opts.MaxBodySize, "maximum size of a single command body")
+	flagSet.Duration("max-defer-timeout", opts.MaxDeferTimeout, "maximum duration when deferring a message")
 
 	// client overridable configuration options
 	flagSet.Duration("max-heartbeat-interval", opts.MaxHeartbeatInterval, "maximum client configurable duration of time between client heartbeats")

nsqd/http.go

@@ -245,7 +245,7 @@ func (s *httpServer) doPUB(w http.ResponseWriter, req *http.Request, ps httprout
 			return nil, http_api.Err{400, "INVALID_DEFER"}
 		}
 		deferred = time.Duration(di) * time.Millisecond
-		if deferred < 0 || deferred > s.nsqd.getOpts().MaxReqTimeout {
+		if deferred < 0 || deferred > s.nsqd.getOpts().MaxDeferTimeout {
 			return nil, http_api.Err{400, "INVALID_DEFER"}
 		}
 	}

nsqd/options.go

@@ -47,12 +47,13 @@ type Options struct {
 	QueueScanDirtyPercent    float64
 
 	// msg and command options
-	MsgTimeout    time.Duration `flag:"msg-timeout"`
-	MaxMsgTimeout time.Duration `flag:"max-msg-timeout"`
-	MaxMsgSize    int64         `flag:"max-msg-size"`
-	MaxBodySize   int64         `flag:"max-body-size"`
-	MaxReqTimeout time.Duration `flag:"max-req-timeout"`
-	ClientTimeout time.Duration
+	MsgTimeout      time.Duration `flag:"msg-timeout"`
+	MaxMsgTimeout   time.Duration `flag:"max-msg-timeout"`
+	MaxMsgSize      int64         `flag:"max-msg-size"`
+	MaxBodySize     int64         `flag:"max-body-size"`
+	MaxReqTimeout   time.Duration `flag:"max-req-timeout"`
+	ClientTimeout   time.Duration
+	MaxDeferTimeout time.Duration `flag:"max-defer-timeout"`
 
 	// client overridable configuration options
 	MaxHeartbeatInterval   time.Duration `flag:"max-heartbeat-interval"`
@@ -151,12 +152,13 @@ func NewOptions() *Options {
 		QueueScanWorkerPoolMax:   4,
 		QueueScanDirtyPercent:    0.25,
 
-		MsgTimeout:    60 * time.Second,
-		MaxMsgTimeout: 15 * time.Minute,
-		MaxMsgSize:    1024 * 1024,
-		MaxBodySize:   5 * 1024 * 1024,
-		MaxReqTimeout: 1 * time.Hour,
-		ClientTimeout: 60 * time.Second,
+		MsgTimeout:      60 * time.Second,
+		MaxMsgTimeout:   15 * time.Minute,
+		MaxMsgSize:      1024 * 1024,
+		MaxBodySize:     5 * 1024 * 1024,
+		MaxReqTimeout:   1 * time.Hour,
+		ClientTimeout:   60 * time.Second,
+		MaxDeferTimeout: 1 * time.Hour,
 
 		MaxHeartbeatInterval:   60 * time.Second,
 		MaxRdyCount:            2500,

nsqd/protocol_v2.go

@@ -916,14 +916,14 @@ func (p *protocolV2) DPUB(client *clientV2, params [][]byte) ([]byte, error) {
 	timeoutMs, err := protocol.ByteToBase10(params[2])
 	if err != nil {
 		return nil, protocol.NewFatalClientErr(err, "E_INVALID",
-			fmt.Sprintf("DPUB could not parse timeout %s", params[2]))
+			fmt.Sprintf("DPUB could not parse defer timeout %s", params[2]))
 	}
 	timeoutDuration := time.Duration(timeoutMs) * time.Millisecond
 
-	if timeoutDuration < 0 || timeoutDuration > p.nsqd.getOpts().MaxReqTimeout {
+	if timeoutDuration < 0 || timeoutDuration > p.nsqd.getOpts().MaxDeferTimeout {
 		return nil, protocol.NewFatalClientErr(nil, "E_INVALID",
-			fmt.Sprintf("DPUB timeout %d out of range 0-%d",
-				timeoutMs, p.nsqd.getOpts().MaxReqTimeout/time.Millisecond))
+			fmt.Sprintf("DPUB defer timeout %d out of range 0-%d",
+				timeoutMs, p.nsqd.getOpts().MaxDeferTimeout/time.Millisecond))
 	}
 
 	bodyLen, err := readLen(client.Reader, client.lenSlice)

nsqd/protocol_v2_test.go

@@ -671,12 +671,12 @@ func TestDPUB(t *testing.T) {
 	test.Equal(t, 1, int(atomic.LoadUint64(&ch.messageCount)))
 
 	// duration out of range
-	nsq.DeferredPublish(topicName, opts.MaxReqTimeout+100*time.Millisecond, make([]byte, 100)).WriteTo(conn)
+	nsq.DeferredPublish(topicName, opts.MaxDeferTimeout+100*time.Millisecond, make([]byte, 100)).WriteTo(conn)
 	resp, _ = nsq.ReadResponse(conn)
 	frameType, data, _ = nsq.UnpackResponse(resp)
 	t.Logf("frameType: %d, data: %s", frameType, data)
 	test.Equal(t, frameTypeError, frameType)
-	test.Equal(t, "E_INVALID DPUB timeout 3600100 out of range 0-3600000", string(data))
+	test.Equal(t, "E_INVALID DPUB defer timeout 3600100 out of range 0-3600000", string(data))
 }
 
 func TestTouch(t *testing.T) {

nsqd/protocol_v2_unixsocket_test.go

@@ -531,12 +531,12 @@ func TestUnixSocketDPUB(t *testing.T) {
 	test.Equal(t, 1, int(atomic.LoadUint64(&ch.messageCount)))
 
 	// duration out of range
-	nsq.DeferredPublish(topicName, opts.MaxReqTimeout+100*time.Millisecond, make([]byte, 100)).WriteTo(conn)
+	nsq.DeferredPublish(topicName, opts.MaxDeferTimeout+100*time.Millisecond, make([]byte, 100)).WriteTo(conn)
 	resp, _ = nsq.ReadResponse(conn)
 	frameType, data, _ = nsq.UnpackResponse(resp)
 	t.Logf("frameType: %d, data: %s", frameType, data)
 	test.Equal(t, frameTypeError, frameType)
-	test.Equal(t, "E_INVALID DPUB timeout 3600100 out of range 0-3600000", string(data))
+	test.Equal(t, "E_INVALID DPUB defer timeout 3600100 out of range 0-3600000", string(data))
 }
 
 func TestUnixSocketTouch(t *testing.T) {