Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mreiferson merged 1 commit into from
Jun 29, 2025
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions apps/nsqd/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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
}
1 change: 1 addition & 0 deletions apps/nsqd/options.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down
2 changes: 1 addition & 1 deletion nsqd/http.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"}
}
}
Expand Down
26 changes: 14 additions & 12 deletions nsqd/options.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"`
Expand Down Expand Up @@ -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,
Expand Down
8 changes: 4 additions & 4 deletions nsqd/protocol_v2.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
4 changes: 2 additions & 2 deletions nsqd/protocol_v2_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down
4 changes: 2 additions & 2 deletions nsqd/protocol_v2_unixsocket_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down
Loading