Repository navigation
[watermill-sql] A missed ack deadline deletes the unprocessed rows of the batch (PostgreSQLQueueOffsetsAdapter) #700
Description
Steps to reproduce
With PostgreSQLQueueSchema and PostgreSQLQueueOffsetsAdapter, one message that misses AckDeadline causes the loss of all later messages in the same batch.
Subscriber.query stops the batch at the first message that is not acked. Then it gives all selected rows (Rows: messageRows) to AckMessageQuery, not only the acked rows. PostgreSQLQueueOffsetsAdapter.AckMessageQuery deletes (DeleteOnAck: true) or marks as acked (DeleteOnAck: false) every offset in params.Rows. The transaction then commits. The rows after the stopped message are never delivered.
Versions: watermill-sql v4.1.5, watermill v1.5.1, pgx v5.9.2, PostgreSQL 16. The code on master is the same.
docker-compose.yml
services:
postgres:
image: postgres:16
environment:
POSTGRES_PASSWORD: postgres
ports:
- "55432:5432"package main
import (
"context"
stdSQL "database/sql"
"fmt"
"sync"
"time"
"github.com/ThreeDotsLabs/watermill"
"github.com/ThreeDotsLabs/watermill-sql/v4/pkg/sql"
"github.com/ThreeDotsLabs/watermill/message"
_ "github.com/jackc/pgx/v5/stdlib"
)
func main() {
db, err := stdSQL.Open("pgx", "postgres://postgres:postgres@localhost:55432/postgres?sslmode=disable")
if err != nil {
panic(err)
}
logger := watermill.NopLogger{}
schema := sql.PostgreSQLQueueSchema{}
offsets := sql.PostgreSQLQueueOffsetsAdapter{DeleteOnAck: true}
pub, err := sql.NewPublisher(sql.BeginnerFromStdSQL(db), sql.PublisherConfig{SchemaAdapter: schema, AutoInitializeSchema: true}, logger)
if err != nil {
panic(err)
}
const topic = "repro"
for i := 1; i <= 10; i++ {
if err := pub.Publish(topic, message.NewMessage(fmt.Sprintf("msg-%d", i), nil)); err != nil {
panic(err)
}
}
ackDeadline := 100 * time.Millisecond
sub, err := sql.NewSubscriber(sql.BeginnerFromStdSQL(db), sql.SubscriberConfig{
SchemaAdapter: schema,
OffsetsAdapter: offsets,
InitializeSchema: true,
AckDeadline: &ackDeadline,
}, logger)
if err != nil {
panic(err)
}
ctx, cancel := context.WithCancel(context.Background())
msgs, err := sub.Subscribe(ctx, topic)
if err != nil {
panic(err)
}
var mu sync.Mutex
received := map[string]int{}
go func() {
for msg := range msgs {
mu.Lock()
received[msg.UUID]++
mu.Unlock()
if msg.UUID == "msg-3" && received[msg.UUID] == 1 {
time.Sleep(time.Second)
}
msg.Ack()
}
}()
time.Sleep(5 * time.Second)
cancel()
_ = sub.Close()
var remaining int
if err := db.QueryRow(`SELECT count(*) FROM "watermill_repro"`).Scan(&remaining); err != nil {
panic(err)
}
mu.Lock()
defer mu.Unlock()
fmt.Println("rows left in the table:", remaining)
for i := 1; i <= 10; i++ {
id := fmt.Sprintf("msg-%d", i)
fmt.Printf("%s delivered %d times\n", id, received[id])
}
}Output:
rows left in the table: 0
msg-1 delivered 1 times
msg-2 delivered 1 times
msg-3 delivered 1 times
msg-4 delivered 0 times
msg-5 delivered 0 times
msg-6 delivered 0 times
msg-7 delivered 0 times
msg-8 delivered 0 times
msg-9 delivered 0 times
msg-10 delivered 0 times
Expected behavior
The ack query changes only the rows up to and including the last acked row (msg-1 and msg-2). msg-3 to msg-10 stay in the table, and the next query delivers them again.
Actual behavior
All 10 rows are deleted. msg-4 to msg-10 are never delivered. msg-3 is not delivered again, although its ack came after the deadline.
Possible solution
Give AckMessageQuery only the acked rows. For example, in Subscriber.query, keep the acked rows in a separate slice and pass that slice as Rows. As a workaround, we wrap PostgreSQLQueueOffsetsAdapter and keep only the rows with row.Offset <= params.LastRow.Offset.
The open PR ThreeDotsLabs/watermill-sql#62 changes the Nack path, and #224 is about a Nack. A missed ack deadline does not return an error from sendMessage, so that change does not prevent this loss.
🤖 Generated with Claude Code
Activity