2
0
mirror of https://github.com/hibiken/asynq.git synced 2026-04-14 22:55:58 +08:00

Compare commits

...

6 Commits

Author SHA1 Message Date
dependabot[bot]
4767d54112 build(deps): bump codecov/codecov-action from 5 to 6
Bumps [codecov/codecov-action](https://github.com/codecov/codecov-action) from 5 to 6.
- [Release notes](https://github.com/codecov/codecov-action/releases)
- [Changelog](https://github.com/codecov/codecov-action/blob/main/CHANGELOG.md)
- [Commits](https://github.com/codecov/codecov-action/compare/v5...v6)

---
updated-dependencies:
- dependency-name: codecov/codecov-action
  dependency-version: '6'
  dependency-type: direct:production
  update-type: version-update:semver-major
...

Signed-off-by: dependabot[bot] <support@github.com>
2026-04-13 23:45:22 +00:00
Mohamed Sohail
23905a286f Merge pull request #1104 from Bahtya/fix/pubsub-connection-leak
fix: close pubsub connection on Subscribe error in CancelationPubSub
2026-04-13 13:58:23 +03:00
Bahtya
5586efeae7 test: add test for CancelationPubSub error path
Add TestCancelationPubSubReceiveError to verify that when
Receive() fails in CancelationPubSub(), an error is returned
and the pubsub connection is not leaked.

This provides test coverage for the pubsub.Close() fix that
was missing in the previous commit.

Bahtya
2026-04-09 20:11:53 +08:00
bahtya
dd3c923f44 fix: close pubsub connection on Subscribe error in CancelationPubSub
When redis.Subscribe succeeds but Receive() fails, the pubsub
connection was not being closed before returning the error. This caused
the subscriber goroutine to leak a Redis connection on each retry
iteration, eventually exhausting the connection pool.

Fixes #1095
2026-04-09 18:15:48 +08:00
Mohamed Sohail
f81c78e68d Merge pull request #1092 from NilPuig/fix/memory-usage-nil-guard
Fix nil panic in memoryUsageCmd Lua script
2026-04-09 10:08:50 +03:00
Nil
2fd155e31d Fix nil guard for MEMORY USAGE in memoryUsageCmd Lua script
MEMORY USAGE returns nil for keys that no longer exist (e.g., expired
or deleted task keys). In Lua, nil is converted to false (a boolean).
The script then attempts arithmetic on this boolean value, causing:

  ERR user_script:30: attempt to perform arithmetic on local 'bytes'
  (a boolean value)

This breaks the /api/queues endpoint in asynqmon, showing "Could not
retrieve queues live data" in the UI.

The fix adds nil guards around all three MEMORY USAGE calls on task
keys, and a divide-by-zero guard on agg_task_sample_size.

Tested in production with Redis 7.2 and asynq v0.25.1 worker.

Fixes #728
Related to #901
2026-02-07 15:13:58 +08:00
4 changed files with 39 additions and 7 deletions

View File

@@ -39,7 +39,7 @@ jobs:
run: go test -run=^$ -bench=. -loglevel=debug ./...
- name: Upload coverage to Codecov
uses: codecov/codecov-action@v5
uses: codecov/codecov-action@v6
build-tool:
strategy:

View File

@@ -264,7 +264,9 @@ for i=1,2 do
if (table.getn(ids) > 0) then
for _, id in ipairs(ids) do
local bytes = redis.call("MEMORY", "USAGE", ARGV[1] .. id)
sample_total = sample_total + bytes
if bytes then
sample_total = sample_total + bytes
end
end
local n = redis.call("LLEN", KEYS[i])
local avg = sample_total / table.getn(ids)
@@ -281,7 +283,9 @@ for i=3,6 do
if (table.getn(ids) > 0) then
for _, id in ipairs(ids) do
local bytes = redis.call("MEMORY", "USAGE", ARGV[1] .. id)
sample_total = sample_total + bytes
if bytes then
sample_total = sample_total + bytes
end
end
local n = redis.call("ZCARD", KEYS[i])
local avg = sample_total / table.getn(ids)
@@ -304,13 +308,17 @@ if table.getn(groups) > 0 then
local ids = redis.call("ZRANGE", group_key, 0, sample_size - 1)
for _, id in ipairs(ids) do
local bytes = redis.call("MEMORY", "USAGE", ARGV[1] .. id)
agg_task_sample_total = agg_task_sample_total + bytes
agg_task_sample_size = agg_task_sample_size + 1
if bytes then
agg_task_sample_total = agg_task_sample_total + bytes
agg_task_sample_size = agg_task_sample_size + 1
end
end
end
end
local avg = agg_task_sample_total / agg_task_sample_size
memusg = memusg + (avg * agg_task_count)
if agg_task_sample_size > 0 then
local avg = agg_task_sample_total / agg_task_sample_size
memusg = memusg + (avg * agg_task_count)
end
end
return memusg
`)

View File

@@ -1488,6 +1488,7 @@ func (r *RDB) CancelationPubSub() (*redis.PubSub, error) {
pubsub := r.client.Subscribe(ctx, base.CancelChannel)
_, err := pubsub.Receive(ctx)
if err != nil {
pubsub.Close()
return nil, errors.E(op, errors.Unknown, fmt.Sprintf("redis pubsub receive error: %v", err))
}
return pubsub, nil

View File

@@ -3274,6 +3274,29 @@ func TestCancelationPubSub(t *testing.T) {
mu.Unlock()
}
func TestCancelationPubSubReceiveError(t *testing.T) {
// Use a client connected to a non-existent Redis server to trigger
// a Receive() error. This verifies that the pubsub connection is
// closed on error, preventing connection leaks.
client := redis.NewClient(&redis.Options{
Addr: "localhost:0", // invalid port — connection will fail
})
r := NewRDB(client)
defer r.Close()
pubsub, err := r.CancelationPubSub()
if err == nil {
// If no error, we must clean up the pubsub.
if pubsub != nil {
pubsub.Close()
}
t.Fatal("(*RDB).CancelationPubSub() expected to return an error when redis is unreachable")
}
if pubsub != nil {
t.Error("(*RDB).CancelationPubSub() expected nil pubsub on error")
}
}
func TestWriteResult(t *testing.T) {
r := setup(t)
defer r.Close()