filer: keep the listing snapshot below later and unfinished event stamps (#12642)

filer: take the listing snapshot as a claim below later and unfinished stamps

A ListEntries snapshot tells the client that every event at or below it is
already in the listing, so the mount drops those events from a directory
build and fences them with the directory floor. The snapshot was a raw
time.Now(), which an event stamp can tie on a coarse clock: on Windows a
mkdir right after a listing of its parent was stamped with the snapshot's
own timestamp, the mount took it as listed, and the new directory stayed
missing from a parent marked cached. A create stamped before the snapshot
but not yet in the store was dropped the same way.

Take the snapshot through the same in-flight claim the idle heartbeat uses:
it stays below every unfinished stamp, and every later stamp lands above it.

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

weed/server/filer_grpc_server.go

@@ -78,7 +78,7 @@ func (fs *FilerServer) ListEntries(req *filer_pb.ListEntriesRequest, stream file
 	includeLastFile := req.InclusiveStartFrom
 	snapshotTsNs := req.SnapshotTsNs
 	if snapshotTsNs == 0 {
-		snapshotTsNs = time.Now().UnixNano()
+		snapshotTsNs = fs.filer.LocalDeliveredThroughTsNs(time.Now().UnixNano())
 	}
 	sentSnapshot := false
 	var listErr error

weed/server/filer_grpc_server_list_snapshot_test.go

@@ -0,0 +1,74 @@
+package weed_server
+
+import (
+	"context"
+	"strconv"
+	"testing"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/metadata"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+)
+
+type captureListStream struct {
+	grpc.ServerStream
+	responses []*filer_pb.ListEntriesResponse
+	trailer   metadata.MD
+}
+
+func (s *captureListStream) Context() context.Context { return context.Background() }
+
+func (s *captureListStream) Send(resp *filer_pb.ListEntriesResponse) error {
+	s.responses = append(s.responses, resp)
+	return nil
+}
+
+func (s *captureListStream) SetTrailer(md metadata.MD) { s.trailer = md }
+
+func (s *captureListStream) snapshotTsNs(t *testing.T) int64 {
+	t.Helper()
+	values := s.trailer.Get(filer_pb.ListSnapshotTsNsTrailerKey)
+	if len(values) != 1 {
+		t.Fatalf("snapshot trailer = %v", values)
+	}
+	tsNs, err := strconv.ParseInt(values[0], 10, 64)
+	if err != nil {
+		t.Fatalf("snapshot trailer %q: %v", values[0], err)
+	}
+	return tsNs
+}
+
+// A listing snapshot tells the client every event at or below it is in the
+// listing, so the client drops those events. It must stay below a create that
+// is stamped but may not be in the store yet, and below every stamp issued
+// after it, or a create landing right after the listing never reaches the
+// client's cache.
+func TestListEntriesSnapshotBelowUnfinishedAndLaterStamps(t *testing.T) {
+	store := newRenameTestStore()
+	store.entries["/dir"] = newDirectoryEntry("/dir", 100)
+	f := newRenameTestFiler(t, store)
+	fs := &FilerServer{filer: f, option: &FilerOption{DirListingLimit: 1000}}
+
+	inFlightTsNs := f.StampMetaLogInflightForTest()
+	stream := &captureListStream{}
+	if err := fs.ListEntries(&filer_pb.ListEntriesRequest{Directory: "/dir"}, stream); err != nil {
+		t.Fatalf("ListEntries: %v", err)
+	}
+	f.DoneMetaLogInflightForTest(inFlightTsNs)
+	snapshotTsNs := stream.snapshotTsNs(t)
+	if snapshotTsNs >= inFlightTsNs {
+		t.Fatalf("snapshot %d covers the unfinished stamp %d", snapshotTsNs, inFlightTsNs)
+	}
+
+	stream = &captureListStream{}
+	if err := fs.ListEntries(&filer_pb.ListEntriesRequest{Directory: "/dir"}, stream); err != nil {
+		t.Fatalf("ListEntries: %v", err)
+	}
+	snapshotTsNs = stream.snapshotTsNs(t)
+	laterTsNs := f.StampMetaLogInflightForTest()
+	f.DoneMetaLogInflightForTest(laterTsNs)
+	if laterTsNs <= snapshotTsNs {
+		t.Fatalf("stamp %d issued after the listing is not above its snapshot %d", laterTsNs, snapshotTsNs)
+	}
+}