Does the delayed message support JetStream WorkQueue

[wf-soft]

I tried to post a delayed message, but it always returned 'message schedules target is invalid'

await using var nc = new NatsClient("nats://localhost:4223");
var js = nc.CreateJetStreamContext();

var streamConfig1 = new StreamConfig(
    name: $"b",
    subjects: new[] { "b.0" }
);
streamConfig1.AllowMsgSchedules = true;
streamConfig1.AllowMsgTTL = true;
streamConfig1.Retention = StreamConfigRetention.Workqueue;
await js.CreateOrUpdateStreamAsync(streamConfig1);


string streamName = $"a_0";
var streamConfig = new StreamConfig(
    name: streamName,
    subjects: new[] { $"a.0.*" }
);
streamConfig.AllowMsgSchedules = true;
streamConfig.AllowMsgTTL = true;
streamConfig.Retention = StreamConfigRetention.Workqueue;
var streamInfo = await js.CreateOrUpdateStreamAsync(streamConfig);
await streamInfo.PurgeAsync(new StreamPurgeRequest() { });

var time = DateTime.Now.AddSeconds(5).ToString("yyyy-MM-ddTHH:mm:ssZ");
Console.WriteLine(time);
var headers = new NatsHeaders
{
    { "Nats-Schedule-Target", "b.0" },
    { "Nats-Schedule", $"@at {time}" },
    { "Nats-Schedule-TTL", "1s" },
};

var res1 = await js.PublishAsync("a.0.0", "a.0.0.1", opts: new NatsJSPubOpts() { }, headers: headers);
res1.EnsureSuccess();
await js.PublishAsync("a.0.1", "a.0.0.2");
await js.PublishAsync("a.0.0", "a.0.0.3");
await js.PublishAsync("a.0.1", "a.0.0.4", opts: new NatsJSPubOpts() { });

[MauriceVanVeen]

Both the schedule and the delayed message need to have subjects on the same stream.

So, if you create a stream that captures both A and B, then it will work.

[wf-soft]

@MauriceVanVeen Great, there are no more exceptions, but the message seems to have not been delayed. He will immediately enter the queue and be consumed

[wf-soft]

string streamName = $"a_0";
var streamConfig = new StreamConfig(
    name: streamName,
    subjects: new[] { "a.0.*", "b.0" }
);
streamConfig.AllowMsgSchedules = true;
streamConfig.AllowMsgTTL = true;
streamConfig.Retention = StreamConfigRetention.Workqueue;
var streamInfo = await js.CreateOrUpdateStreamAsync(streamConfig);
await streamInfo.PurgeAsync(new StreamPurgeRequest() { });

var time = DateTime.Now.AddSeconds(5).ToString("yyyy-MM-ddTHH:mm:ssZ");
Console.WriteLine(time);
var headers = new NatsHeaders
{
    { "Nats-Schedule-Target", "b.0" },
    { "Nats-Schedule", $"@at {time}" },
    //{ "Nats-Schedule-TTL", "5s" },

};

var res1 = await js.PublishAsync("a.0.0", "a.0.0.1", opts: new NatsJSPubOpts() { }, headers: headers);
res1.EnsureSuccess();
await js.PublishAsync("a.0.1", "a.0.0.2");
await js.PublishAsync("a.0.0", "a.0.0.3");
await js.PublishAsync("a.0.1", "a.0.0.4", opts: new NatsJSPubOpts() { });

var consumer = await js.CreateOrUpdateConsumerAsync("a_0", new ConsumerConfig("parse")
{
    DurableName = "parse"
});

for (int i = 0; i < 4; i++)
{
    var msg = await consumer.NextAsync<string>(opts: new NatsJSNextOpts() { Expires = TimeSpan.FromSeconds(10) });
    if (msg is not null)
    {
        Console.WriteLine(msg.Value.Data);
        await msg.Value.AckAsync();
    }
}

I got this output

a.0.0.1
a.0.0.2
a.0.0.3
a.0.0.4

[MauriceVanVeen]

The delayed message will be scheduled on the target subject of b.0. That will take 5 seconds in this case. You'll likely need to iterate and request at least one more message.

[wf-soft]

@MauriceVanVeen I still haven't received any message from subject b.0

for (int i = 0; i < 5; i++)
{
    var msg = await consumer.NextAsync<string>(opts: new NatsJSNextOpts() { Expires = TimeSpan.FromSeconds(20) });
    if (msg is not null)
    {
        Console.WriteLine(msg.Value.Data);
        await msg.Value.AckAsync();
    }
    else Console.WriteLine("null");
}
Console.WriteLine("exit");

a.0.0.1
a.0.0.2
a.0.0.3
a.0.0.4
null
exit

[MauriceVanVeen]

Seems there are a couple issues:

• This line: var time = DateTime.Now.AddSeconds(5).ToString("yyyy-MM-ddTHH:mm:ssZ"); converts the current time as-if it is in UTC. Depending on your timezone offset, this could mean it would takes hours for this schedule to fire, not just 5 seconds. Should use DateTime.Now.AddSeconds(5).ToString("o") instead.
• You've made the stream WorkQueue, and once you consumed and Acked the message with subject a.0.0 it removed the schedule early! Converting the stream to limits based is an easy fix for this. Or, generally you'd want a consumer that filters on and only receives the delayed messages and not also the schedules themselves.

Additionally, a subject can only hold one schedule. So if you have multiple values for a subject, for example here you're posting more values to the same subject a.0.0, then only the last will hold the active schedule. The delayed message still gets successfully scheduled, but this seems like a bug honestly because the second publish to a.0.0 should deactivate the original schedule, so will fix that for 2.12.1+.

[wf-soft]

@MauriceVanVeen Okay, currently using Limits Stream will consume delayed messages twice. Looking forward to the next version

[wf-soft]

@MauriceVanVeen I still hope that Workqueue will receive more support, as Workqueue is more suitable for most scenarios

[MauriceVanVeen]

You'll only receive the delayed message once, but you also received the schedule immediately. The schedule is in one message that you can consume, and the delayed message is another message.
If you don't want to receive it twice, you should put a filter subject on the consumer such that you only receive delayed messages under subject b.*.

Importantly, the retention policy still applies. WorkQueue is a supported configuration, but that means when a message Is acknowledged it is deleted under the WorkQueue policy. The schedule is just another message that you could ack and delete like this (which means the schedule would be deleted).
Using a filter subject like mentioned above will also fix it for your use case.