diff --git a/apps/nsqd/main.go b/apps/nsqd/main.go index c9a6b9bd3..a5cf2aee0 100644 --- a/apps/nsqd/main.go +++ b/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 +} diff --git a/apps/nsqd/options.go b/apps/nsqd/options.go index f93597013..5d3942ef7 100644 --- a/apps/nsqd/options.go +++ b/apps/nsqd/options.go @@ -159,6 +159,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") diff --git a/nsqd/http.go b/nsqd/http.go index c67424187..b25e41ec9 100644 --- a/nsqd/http.go +++ b/nsqd/http.go @@ -241,7 +241,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"} } } diff --git a/nsqd/options.go b/nsqd/options.go index 8997c7265..8420b7c70 100644 --- a/nsqd/options.go +++ b/nsqd/options.go @@ -45,12 +45,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"` @@ -127,12 +128,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, diff --git a/nsqd/protocol_v2.go b/nsqd/protocol_v2.go index 8ec422430..9696868c9 100644 --- a/nsqd/protocol_v2.go +++ b/nsqd/protocol_v2.go @@ -881,14 +881,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) diff --git a/nsqd/protocol_v2_test.go b/nsqd/protocol_v2_test.go index 19015d1e9..2064b12ad 100644 --- a/nsqd/protocol_v2_test.go +++ b/nsqd/protocol_v2_test.go @@ -596,12 +596,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) { diff --git a/nsqd/protocol_v2_unixsocket_test.go b/nsqd/protocol_v2_unixsocket_test.go index 85820aca6..fa5af1fe9 100644 --- a/nsqd/protocol_v2_unixsocket_test.go +++ b/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) {