Direct Get multi last for request returns messages after up_to_time

[scottf]

Behavior

Direct get multi last for with up_to_time returns message after up_to_time

Server version

v2.11.0-dev (Head of main on 05-20-2025)

Steps to reproduce

1. Create a new stream.
2. Publish some messages.
3. get some messages directly to know their times
4. Make some multi-last requests for subjects using a time before the first message time.
   i.e. {"multi_last":["subject6077"],"up_to_time":"2024-11-05T13:31:29.144000000Z"} and see unexpected message.

Comments

It's possible that making the request in this manner is not supported, or that the expectation to get no messages back is incorrect.

[ljluestc]

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"sync"
	"time"

	"github.com/google/uuid"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
	"nats-direct-get-bug/models"
)

func main() {
	// Connect to NATS
	nc, err := nats.Connect("nats://localhost:4222")
	if err != nil {
		log.Fatalf("Failed to connect to NATS: %v", err)
	}
	defer nc.Close()

	// Create JetStream context
	js, err := jetstream.New(nc)
	if err != nil {
		log.Fatalf("Failed to create JetStream context: %v", err)
	}

	// Create stream
	ctx := context.Background()
	stream, err := js.CreateStream(ctx, jetstream.StreamConfig{
		Name:     "TEST",
		Subjects: []string{"testSubject"},
		Storage:  jetstream.FileStorage,
	})
	if err != nil {
		log.Fatalf("Failed to create stream: %v", err)
	}

	// Publish one message
	msg := []byte("Test Message")
	pubAck, err := stream.Publish(ctx, "testSubject", msg)
	if err != nil {
		log.Fatalf("Failed to publish message: %v", err)
	}
	msgTime := pubAck.StreamSequenceInfo.Timestamp
	fmt.Printf("Published message at %s\n", msgTime.Format(time.RFC3339Nano))

	// Subscribe to reply subject
	replySubject := fmt.Sprintf("reply.%s", uuid.New().String())
	sub, err := nc.SubscribeSync(replySubject)
	if err != nil {
		log.Fatalf("Failed to subscribe: %v", err)
	}
	defer sub.Unsubscribe()

	// Create direct get request with up_to_time
	upToTime := msgTime.Add(10 * time.Second) // 10 seconds after message time
	req := map[string]interface{}{
		"multi_last":   []string{"testSubject"},
		"up_to_time":   upToTime.Format(time.RFC3339Nano),
	}
	reqBytes, err := json.Marshal(req)
	if err != nil {
		log.Fatalf("Failed to marshal request: %v", err)
	}

	// Send direct get request
	if err := nc.PublishRequest("$JS.API.DIRECT.GET.TEST", replySubject, reqBytes); err != nil {
		log.Fatalf("Failed to publish direct get request: %v", err)
	}

	// Collect response
	msg, err = sub.NextMsg(2 * time.Second)
	if err != nil {
		log.Fatalf("Failed to receive response: %v", err)
	}

	var resp map[string]interface{}
	if err := json.Unmarshal(msg.Data, &resp); err != nil {
		log.Fatalf("Failed to unmarshal response: %v", err)
	}
	fmt.Printf("Direct get response: %+v\n", resp)

	// Scatter-gather for user data
	scatterGatherUsers(nc, js)
}

func scatterGatherUsers(nc *nats.Conn, js jetstream.JetStream) {
	// Create or get USERS stream
	ctx := context.Background()
	stream, err := js.CreateStream(ctx, jetstream.StreamConfig{
		Name:     "USERS",
		Subjects: []string{"users.*"},
		Storage:  jetstream.FileStorage,
	})
	if err != nil {
		log.Fatalf("Failed to create USERS stream: %v", err)
	}

	// Send scatter-gather request
	requestID := uuid.New().String()
	req := models.OnlineUsersRequest{RequestID: requestID}
	reqBytes, err := json.Marshal(req)
	if err != nil {
		log.Fatalf("Error marshaling request: %v", err)
	}

	var mu sync.Mutex
	responses := make([]models.OnlineUsersResponse, 0)
	timeout := 2 * time.Second

	replySubject := fmt.Sprintf("reply.%s", requestID)
	sub, err := nc.SubscribeSync(replySubject)
	if err != nil {
		log.Fatalf("Failed to subscribe: %v", err)
	}
	defer sub.Unsubscribe()

	if err := nc.PublishRequest("get_online_users", replySubject, reqBytes); err != nil {
		log.Fatalf("Failed to publish request: %v", err)
	}

	start := time.Now()
	for time.Since(start) < timeout {
		msg, err := sub.NextMsg(timeout - time.Since(start))
		if err == nats.ErrTimeout {
			break
		}
		if err != nil {
			log.Printf("Error receiving message: %v", err)
			continue
		}

		var resp models.OnlineUsersResponse
		if err := json.Unmarshal(msg.Data, &resp); err != nil {
			log.Printf("Error unmarshaling response: %v", err)
			continue
		}

		mu.Lock()
		responses = append(responses, resp)
		mu.Unlock()
	}

	allUsers := make([]models.User, 0)
	for _, resp := range responses {
		fmt.Printf("Service %s returned %d users\n", resp.ServiceID, len(resp.Users))
		allUsers = append(allUsers, resp.Users...)

		userData, _ := json.Marshal(resp.Users)
		_, err := stream.Publish(ctx, fmt.Sprintf("users.%s", resp.ServiceID), userData)
		if err != nil {
			log.Printf("Failed to publish to JetStream: %v", err)
		}
	}

	fmt.Printf("Total users: %d\n", len(allUsers))
	for _, user := range allUsers {
		fmt.Printf("User ID: %s, Name: %s\n", user.ID, user.Name)
	}
}

[scottf]

I can still reproduce this issue against Server. This code uses the Java Orbit DirectBatch library

String stream = "stream6077";
String subject = "subject6077";
try (Connection nc = Nats.connect(runner.getURI())) {
    JetStreamManagement jsm = nc.jetStreamManagement();
    JetStream js = jsm.jetStream();

    // 1. create a new stream
    jsm.addStream(StreamConfiguration.builder()
        .name(stream)
        .subjects(subject)
        .allowDirect(true)
        .storageType(StorageType.Memory)
        .build());

    // 2. Publish some messages.
    for (int x = 1; x <= 10; x++) {
        js.publish(subject, ("Message " + x).getBytes());
        Thread.sleep(100);
    }

    // 3. get some messages directly to know their times
    MessageInfo mi1 = jsm.getMessage(stream, 1);
    MessageInfo mi5 = jsm.getMessage(stream, 5);

    System.out.println(mi1);
    System.out.println(mi5);

    // 4. Use the Direct Batch Context to make some multi-last requests for subjects and time
    // (Direct Batch Context is an Orbit extension)
    DirectBatchContext db = new DirectBatchContext(nc, stream);

    MessageBatchGetRequest request = MessageBatchGetRequest.multiLastForSubjects(Collections.singletonList(subject), mi5.getTime());
    System.out.println("Request against time of message sequence 5:\n    " + request);
    List<MessageInfo> list = db.fetchMessageBatch(request);
    for (MessageInfo mi : list) {
        System.out.println("    " + mi);
    }

    request = MessageBatchGetRequest.multiLastForSubjects(Collections.singletonList(subject), mi5.getTime());
    System.out.println("Request against time of message sequence 1:\n    " + request);
    list = db.fetchMessageBatch(request);
    for (MessageInfo mi : list) {
        System.out.println("    " + mi);
    }
}

Here is the output (manually formatted)

Direct Get MessageInfo for sequence 1:

{
    "seq": 1,
    "num_pending": -1,
    "stream": "stream6077",
    "subject": "subject6077",
    "time": "2025-05-20T17:53:03.458042100Z",
    "data_length": 9
}

Direct Get MessageInfo for sequence 5:

{
    "seq": 5,
    "num_pending": -1,
    "stream": "stream6077",
    "subject": "subject6077",
    "time": "2025-05-20T17:53:03.897816100Z",
    "data_length": 9
}

Request against time of message sequence 5:

{"multi_last":["subject6077"],"up_to_time":"2025-05-20T17:53:03.897816100Z"}

Response. 1 message as expected.

{
    "seq": 4,
    "last_seq": 0,
    "num_pending": -1,
    "stream": "stream6077",
    "subject": "subject6077",
    "time": "2025-05-20T17:53:03.787923000Z",
    "data_length": 9
}

Request against time of message sequence 1:

{"multi_last":["subject6077"],"up_to_time":"2025-05-20T17:53:03.458042100Z"}

Response. 1 message not as expected

{
    "seq": 10,
    "last_seq": 0,
    "num_pending": -1,
    "stream": "stream6077",
    "subject": "subject6077",
    "time": "2025-05-20T17:53:04.443040400Z",
    "data_length": 10
}

[scottf]

@ljluestc Hey I saw your code sample, I wondering what it was in relation too. Were you adding code to reproduce the issue? Go is not my first language and I wasn't sure what the scatterGatherUsers portion was for.

[neilalexander]

I have reproduced this in a server test and will post a fix soon.