client: HTTP timeout must be larger than the consume timeout (#233)

Author: calvinxiao

client/client.go

@@ -99,6 +99,45 @@ func (c *LmstfyClient) ConfigRetry(retryCount int, backOffMillisecond int) {
 	c.backOff = backOffMillisecond
 }
 
+// validateConsumeTimeout validates that the consume timeout is less than HTTP client timeout
+func (c *LmstfyClient) validateConsumeTimeout(timeoutSecond uint32) *APIError {
+	if timeoutSecond == 0 {
+		return nil
+	}
+
+	httpTimeout := c.getHTTPTimeout()
+	if httpTimeout <= time.Duration(timeoutSecond)*time.Second {
+		return &APIError{
+			Type: RequestErr,
+			Reason: fmt.Sprintf("consume timeout (%d seconds) must be less than HTTP client timeout (%d seconds)",
+				timeoutSecond, int(httpTimeout.Seconds())),
+		}
+	}
+	return nil
+}
+
+// getHTTPTimeout returns the HTTP client timeout duration
+func (c *LmstfyClient) getHTTPTimeout() time.Duration {
+	if c.httpCli == nil {
+		return maxReadTimeout * time.Second
+	}
+
+	// Check the client timeout
+	if c.httpCli.Timeout > 0 {
+		return c.httpCli.Timeout
+	}
+
+	// Try to get the timeout from the transport and client
+	if transport, ok := c.httpCli.Transport.(*http.Transport); ok {
+		if transport.ResponseHeaderTimeout > 0 {
+			return transport.ResponseHeaderTimeout
+		}
+	}
+
+	// Default to maxReadTimeout
+	return maxReadTimeout * time.Second
+}
+
 func (c *LmstfyClient) getReq(method, relativePath string, query url.Values, body []byte) (req *http.Request, err error) {
 	targetUrl := url.URL{
 		Scheme:   c.scheme,
@@ -242,7 +281,9 @@ RETRY:
 //   - ttlSecond is the time-to-live of the job. If it's zero, job won't expire; if it's positive, the value is the TTL.
 //   - tries is the maximum times the job can be fetched.
 //   - delaySecond is the duration before the job is released for consuming. When it's zero, no delay is applied.
-func (c *LmstfyClient) BatchPublish(queue string, jobs []interface{}, ttlSecond uint32, tries uint16, delaySecond uint32) (jobIDs []string, e error) {
+func (c *LmstfyClient) BatchPublish(queue string, jobs []interface{}, ttlSecond uint32,
+	tries uint16, delaySecond uint32,
+) (jobIDs []string, e error) {
 	query := url.Values{}
 	query.Add("ttl", strconv.FormatUint(uint64(ttlSecond), 10))
 	query.Add("tries", strconv.FormatUint(uint64(tries), 10))
@@ -363,6 +404,11 @@ func (c *LmstfyClient) consume(queue string, ttrSecond, timeoutSecond uint32, fr
 			Reason: fmt.Sprintf("timeout should be < %d", maxReadTimeout),
 		}
 	}
+
+	// Check if HTTP client timeout is larger than consume timeout
+	if err := c.validateConsumeTimeout(timeoutSecond); err != nil {
+		return nil, err
+	}
 	query := url.Values{}
 	query.Add("ttr", strconv.FormatUint(uint64(ttrSecond), 10))
 	query.Add("timeout", strconv.FormatUint(uint64(timeoutSecond), 10))
@@ -436,7 +482,9 @@ func (c *LmstfyClient) BatchConsume(queues []string, count, ttrSecond, timeoutSe
 //     these job will be released for consuming again if the `(tries - 1) > 0`.
 //   - count is the job count of this consume. If it's zero or over 100, this method will return an error.
 //     If it's positive, this method would return some jobs, and it's count is between 0 and count.
-func (c *LmstfyClient) BatchConsumeWithFreezeTries(queues []string, count, ttrSecond, timeoutSecond uint32) (jobs []*Job, e error) {
+func (c *LmstfyClient) BatchConsumeWithFreezeTries(queues []string,
+	count, ttrSecond, timeoutSecond uint32,
+) (jobs []*Job, e error) {
 	return c.batchConsume(queues, count, ttrSecond, timeoutSecond, true)
 }
 
@@ -473,6 +521,11 @@ func (c *LmstfyClient) batchConsume(queues []string, count, ttrSecond, timeoutSe
 		}
 	}
 
+	// Check if HTTP client timeout is larger than consume timeout
+	if err := c.validateConsumeTimeout(timeoutSecond); err != nil {
+		return nil, err
+	}
+
 	query := url.Values{}
 	query.Add("ttr", strconv.FormatUint(uint64(ttrSecond), 10))
 	query.Add("count", strconv.FormatUint(uint64(count), 10))
@@ -569,6 +622,11 @@ func (c *LmstfyClient) consumeFromQueues(ttrSecond, timeoutSecond uint32, freeze
 			Reason: fmt.Sprintf("timeout must be < %d when fetch from multiple queues", maxReadTimeout),
 		}
 	}
+
+	// Check if HTTP client timeout is larger than consume timeout
+	if err := c.validateConsumeTimeout(timeoutSecond); err != nil {
+		return nil, err
+	}
 	query := url.Values{}
 	query.Add("ttr", strconv.FormatUint(uint64(ttrSecond), 10))
 	query.Add("timeout", strconv.FormatUint(uint64(timeoutSecond), 10))

client/client_test.go

@@ -3,6 +3,8 @@ package client
 import (
 	"bytes"
 	"encoding/json"
+	"errors"
+	"net/http"
 	"testing"
 	"time"
 
@@ -307,3 +309,138 @@ func TestLmstfyClient_DeleteDeadLetter(t *testing.T) {
 		t.Fatal("delete deadletter failed")
 	}
 }
+
+func TestLmstfyClient_ValidateConsumeTimeout(t *testing.T) {
+	// Test with default HTTP client (600 seconds timeout)
+	cli := NewLmstfyClient(Host, Port, Namespace, Token)
+
+	// Test valid timeout (should return nil)
+	err := cli.validateConsumeTimeout(300)
+	if err != nil {
+		t.Fatalf("Expected nil for valid timeout, got: %v", err)
+	}
+
+	// Test zero timeout (should return nil)
+	err = cli.validateConsumeTimeout(0)
+	if err != nil {
+		t.Fatalf("Expected nil for zero timeout, got: %v", err)
+	}
+
+	// Test invalid timeout (should return error)
+	err = cli.validateConsumeTimeout(600)
+	if err == nil {
+		t.Fatal("Expected error for timeout >= HTTP timeout")
+	}
+	expectedMsg := "consume timeout (600 seconds) must be less than HTTP client timeout (600 seconds)"
+	var apiErr *APIError
+	if !errors.As(err, &apiErr) || apiErr.Reason != expectedMsg {
+		t.Fatalf("Expected error message '%s', got '%v'", expectedMsg, err)
+	}
+}
+
+func TestLmstfyClient_ConsumeWithTimeoutValidation(t *testing.T) {
+	// Test with custom HTTP client with short timeout
+	shortHTTPClient := &http.Client{
+		Timeout: 5 * time.Second,
+	}
+	cli := NewLmstfyWithClient(shortHTTPClient, Host, Port, Namespace, Token)
+
+	// Test consume with valid timeout
+	job, err := cli.Consume("test-timeout-validation", 10, 3)
+	if err != nil {
+		t.Fatalf("Consume should succeed with valid timeout: %v", err)
+	}
+	if job != nil {
+		cli.Ack("test-timeout-validation", job.ID)
+	}
+
+	// Test consume with invalid timeout (should fail)
+	_, err = cli.Consume("test-timeout-validation", 10, 6)
+	if err == nil {
+		t.Fatal("Consume should fail with timeout >= HTTP timeout")
+	}
+	expectedMsg := "consume timeout (6 seconds) must be less than HTTP client timeout (5 seconds)"
+	var apiErr *APIError
+	if !errors.As(err, &apiErr) || apiErr.Reason != expectedMsg {
+		t.Fatalf("Expected error message '%s', got '%v'", expectedMsg, err)
+	}
+}
+
+func TestLmstfyClient_BatchConsumeWithTimeoutValidation(t *testing.T) {
+	// Test with custom HTTP client with short timeout
+	shortHTTPClient := &http.Client{
+		Timeout: 5 * time.Second,
+	}
+	cli := NewLmstfyWithClient(shortHTTPClient, Host, Port, Namespace, Token)
+
+	// Test batch consume with valid timeout
+	queues := []string{"test-batch-timeout-validation"}
+	jobs, err := cli.BatchConsume(queues, 3, 10, 3)
+	if err != nil {
+		t.Fatalf("BatchConsume should succeed with valid timeout: %v", err)
+	}
+
+	// Ack any jobs that were returned
+	for _, job := range jobs {
+		cli.Ack("test-batch-timeout-validation", job.ID)
+	}
+
+	// Test batch consume with invalid timeout (should fail)
+	_, err = cli.BatchConsume(queues, 3, 10, 6)
+	if err == nil {
+		t.Fatal("BatchConsume should fail with timeout >= HTTP timeout")
+	}
+	expectedMsg := "consume timeout (6 seconds) must be less than HTTP client timeout (5 seconds)"
+	var apiErr *APIError
+	if !errors.As(err, &apiErr) || apiErr.Reason != expectedMsg {
+		t.Fatalf("Expected error message '%s', got '%v'", expectedMsg, err)
+	}
+}
+
+func TestLmstfyClient_ConsumeFromQueuesWithTimeoutValidation(t *testing.T) {
+	// Test with custom HTTP client with short timeout
+	shortHTTPClient := &http.Client{
+		Timeout: 5 * time.Second,
+	}
+	cli := NewLmstfyWithClient(shortHTTPClient, Host, Port, Namespace, Token)
+
+	// Test consume from queues with valid timeout
+	job, err := cli.ConsumeFromQueues(10, 3, "test-multi-queue-timeout-validation")
+	if err != nil {
+		t.Fatalf("ConsumeFromQueues should succeed with valid timeout: %v", err)
+	}
+	if job != nil {
+		cli.Ack("test-multi-queue-timeout-validation", job.ID)
+	}
+
+	// Test consume from queues with invalid timeout (should fail)
+	_, err = cli.ConsumeFromQueues(10, 6, "test-multi-queue-timeout-validation")
+	if err == nil {
+		t.Fatal("ConsumeFromQueues should fail with timeout >= HTTP timeout")
+	}
+	expectedMsg := "consume timeout (6 seconds) must be less than HTTP client timeout (5 seconds)"
+	var apiErr *APIError
+	if !errors.As(err, &apiErr) || apiErr.Reason != expectedMsg {
+		t.Fatalf("Expected error message '%s', got '%v'", expectedMsg, err)
+	}
+}
+
+func TestLmstfyClient_GetHTTPTimeout(t *testing.T) {
+	// Test with default HTTP client
+	cli := NewLmstfyClient(Host, Port, Namespace, Token)
+	timeout := cli.getHTTPTimeout()
+	if timeout != 600*time.Second {
+		t.Fatalf("Expected 600 seconds timeout for default client, got %v", timeout)
+	}
+
+	// Test with custom HTTP client
+	customTimeout := 30 * time.Second
+	customClient := &http.Client{
+		Timeout: customTimeout,
+	}
+	cli2 := NewLmstfyWithClient(customClient, Host, Port, Namespace, Token)
+	timeout = cli2.getHTTPTimeout()
+	if timeout != customTimeout {
+		t.Fatalf("Expected %v timeout for custom client, got %v", customTimeout, timeout)
+	}
+}

engine/redis/queue.go

@@ -115,7 +115,9 @@ func (q *Queue) Destroy() (count int64, err error) {
 	poolPrefix := PoolJobKeyPrefix(q.name.Namespace, q.name.Queue)
 	var batchSize int64 = 100
 	for {
-		val, err := q.redis.Conn.EvalSha(dummyCtx, q.destroySHA, []string{q.Name(), poolPrefix}, batchSize).Result()
+		val, err := q.redis.Conn.EvalSha(dummyCtx, q.destroySHA, []string{
+			q.Name(), poolPrefix,
+		}, batchSize).Result()
 		if err != nil {
 			if isLuaScriptGone(err) {
 				if err := PreloadDeadLetterLuaScript(q.redis); err != nil {
@@ -174,7 +176,9 @@ func popMultiQueues(redis *RedisInstance, queueNames []string) (string, string,
 }
 
 // Poll from multiple queues using blocking method; OR pop a job from one queue using non-blocking method
-func PollQueues(redis *RedisInstance, timer *Timer, queueNames []QueueName, timeoutSecond, ttrSecond uint32) (queueName *QueueName, jobID string, retries uint16, err error) {
+func PollQueues(redis *RedisInstance, timer *Timer, queueNames []QueueName,
+	timeoutSecond, ttrSecond uint32,
+) (queueName *QueueName, jobID string, retries uint16, err error) {
 	defer func() {
 		if jobID != "" {
 			metrics.queuePopJobs.WithLabelValues(redis.Name).Inc()
@@ -236,7 +240,8 @@ func PollQueues(redis *RedisInstance, timer *Timer, queueNames []QueueName, time
 }
 
 // Pack (tries, jobID) into lua struct pack of format "HHHc0", in lua this can be done:
-//   ```local data = struct.pack("HHc0", tries, #job_id, job_id)```
+//
+//	```local data = struct.pack("HHc0", tries, #job_id, job_id)```
 func structPack(tries uint16, jobID string) (data string) {
 	buf := make([]byte, 2+2+len(jobID))
 	binary.LittleEndian.PutUint16(buf[0:], tries)
@@ -246,7 +251,8 @@ func structPack(tries uint16, jobID string) (data string) {
 }
 
 // Unpack the "HHc0" lua struct format, in lua this can be done:
-//   ```local tries, job_id = struct.unpack("HHc0", data)```
+//
+//	```local tries, job_id = struct.unpack("HHc0", data)```
 func structUnpack(data string) (tries uint16, jobID string, err error) {
 	buf := []byte(data)
 	h1 := binary.LittleEndian.Uint16(buf[0:])