diff --git a/weed/filer/filer_notify.go b/weed/filer/filer_notify.go index 7c16107dbe..e71a8b90a1 100644 --- a/weed/filer/filer_notify.go +++ b/weed/filer/filer_notify.go @@ -204,9 +204,10 @@ func (f *Filer) LocalFlushedThroughTsNs(nowNs int64) int64 { } // LocalDeliveredThroughTsNs caps a delivery-freshness claim (an idle -// heartbeat's timestamp) by the oldest in-flight stamp: a stamped-but- -// unappended event has not been streamed to anyone, and a peer aggregator -// turns the claim into its delivery low-watermark. +// heartbeat's timestamp, a listing snapshot) by the oldest in-flight stamp: a +// stamped-but-unappended event has not been streamed to anyone, nor is its +// write certain to be visible yet, and a peer aggregator turns the claim into +// its delivery low-watermark. func (f *Filer) LocalDeliveredThroughTsNs(nowNs int64) int64 { return f.metaLogInflight.claimThrough(nowNs) } diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go index eb3b455b79..9b8808f192 100644 --- a/weed/server/filer_grpc_server.go +++ b/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 diff --git a/weed/server/filer_grpc_server_list_snapshot_test.go b/weed/server/filer_grpc_server_list_snapshot_test.go new file mode 100644 index 0000000000..6269388bc2 --- /dev/null +++ b/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) + } +}