nats driver jobs ack error

[shellphy]

when i use delete_after_ack: true

error message:

2024-11-15 17:39:01.941 ERROR jobs [email protected]/listener.go:162 response handler error {"error": "jobs_handle_response: nats: message deletion unsuccessful: %!s()", "ID": "9435f2a7-7ed9-401c-88de-708ce50faa1c", "response": "{"type":2,"data":[]}", "start": "2024-11-15 17:38:58.773 ", "elapsed": 3168}

./nats-server --version
nats-server: v2.10.22

[rustatian]

Hey @shellphy 👋
What is the question?

[shellphy]

You can try this:

worker.php:

<?php

/**
 * @var Goridge\RelayInterface $relay
 */

use Spiral\Goridge;
use Spiral\RoadRunner;
use Spiral\Goridge\StreamRelay;
use Spiral\RoadRunner\Jobs\Consumer;

ini_set('display_errors', 'stderr');
require "vendor/autoload.php";

$consumer = new Spiral\RoadRunner\Jobs\Consumer();

while ($task = $consumer->waitTask()) {
    try {
        sleep(35);
        $task->ack();
    } catch (\Throwable $e) {
        var_dump($e->getMessage());
    }
}

producer.php:

<?php

/**
 * @var Goridge\RelayInterface $relay
 */

use Spiral\Goridge;
use Spiral\RoadRunner;
use Spiral\Goridge\StreamRelay;
use Spiral\RoadRunner\Jobs\Consumer;

ini_set('display_errors', 'stderr');
require "vendor/autoload.php";

$rpc = new Goridge\RPC\RPC(
    Goridge\Relay::create('tcp://127.0.0.1:6001')
);
$jobs = new Spiral\RoadRunner\Jobs\Jobs($rpc);
$queue = $jobs->connect('test');
$task = $queue->create("dummy", "hello");
$queue->dispatch($task);

.rr.yaml:

version: '3'

rpc:
  listen: tcp://127.0.0.1:6001

server:
  command: "php worker.php"
  relay: "pipes"
  relay_timeout: "20s"

nats:
  addr: "nats://127.0.0.1:4222"

jobs:
  consume: ["test"]
  pool:
    num_workers: 2
  pipelines:
    test:
      driver: nats
      config:
        prefetch: 1
        subject: test
        stream: test
        delete_after_ack: true

logs:
  level: debug
  encoding: console
  mode: development

start nats:

/opt/homebrew/opt/nats-server/bin/nats-server -js

start rr server:

./rr serve

produce message:

php producer.php

output log:

2024-11-15T18:34:24+0000        DEBUG   rpc             plugin was started      {"address": "tcp://127.0.0.1:6001", "list of the plugins with RPC methods:": ["lock", "jobs", "app", "resetter", "informer"]}
2024-11-15T18:34:24+0000        DEBUG   jobs            initializing driver     {"pipeline": "test", "driver": "nats"}
2024-11-15T18:34:24+0000        DEBUG   jobs            driver ready    {"pipeline": "test", "driver": "nats", "start": "2024-11-15T18:34:24+0000", "elapsed": 2}
2024-11-15T18:34:24+0000        DEBUG   nats            pipeline was started    {"driver": "nats", "pipeline": "test", "start": "2024-11-15T18:34:24+0000", "elapsed": 1}
2024-11-15T18:34:24+0000        DEBUG   server          worker is allocated     {"pid": 56770, "max_execs": 0, "internal_event_name": "EventWorkerConstruct"}
2024-11-15T18:34:24+0000        DEBUG   server          worker is allocated     {"pid": 56769, "max_execs": 0, "internal_event_name": "EventWorkerConstruct"}
[INFO] RoadRunner server started; version: 2024.2.1, buildtime: 2024-09-12T16:25:03+0000
[INFO] sdnotify: not notified
2024-11-15T18:34:25+0000        DEBUG   jobs            job was pushed successfully     {"ID": "ebeb0ac6-8fd2-40b8-b212-f8699d086768", "pipeline": "test", "driver": "nats", "start": "2024-11-15T18:34:25+0000", "elapsed": 1}
2024-11-15T18:34:25+0000        DEBUG   jobs            job processing was started      {"ID": "ebeb0ac6-8fd2-40b8-b212-f8699d086768", "start": "2024-11-15T18:34:24+0000", "elapsed": 1581}
2024-11-15T18:34:55+0000        DEBUG   jobs            job processing was started      {"ID": "ebeb0ac6-8fd2-40b8-b212-f8699d086768", "start": "2024-11-15T18:34:24+0000", "elapsed": 31585}
2024-11-15T18:35:00+0000        DEBUG   server          req-resp mode   {"pid": 56770}
2024-11-15T18:35:00+0000        DEBUG   jobs            job was processed successfully  {"ID": "ebeb0ac6-8fd2-40b8-b212-f8699d086768", "start": "2024-11-15T18:34:24+0000", "elapsed": 36591}
2024-11-15T18:35:30+0000        DEBUG   server          req-resp mode   {"pid": 56769}
2024-11-15T18:35:30+0000        ERROR   jobs            response handler error  {"error": "jobs_handle_response: nats: message deletion unsuccessful: %!s(<nil>)", "ID": "ebeb0ac6-8fd2-40b8-b212-f8699d086768", "response": "{\"type\":2,\"data\":[]}", "start": "2024-11-15T18:34:24+0000", "elapsed": 66599}
2024-11-15T18:35:30+0000        ERROR   jobs            acknowledge failed, job might be lost   {"ID": "ebeb0ac6-8fd2-40b8-b212-f8699d086768", "error": "jobs_handle_response: nats: message deletion unsuccessful: %!s(<nil>)", "error": "nats: message was already acknowledged"}

explain

2024-11-15T18:34:25+0000: job started

2024-11-15T18:34:55+0000: The heartbeat is not sent regularly, so after 30 seconds, the job second started

2024-11-15T18:35:00+0000: after 35 seconds, job send ack, nats delete the message

2024-11-15T18:35:30+0000: acknowledge failed, job might be lost