fix(volume): persist the leveldb needle map watermark at batch boundaries (#10557)

levelDbWrite persists the replay watermark when its updateWatermark
argument is true. Put and Delete passed "watermark == 0", which is true
on exactly the writes that carry no checkpoint and false on the batch
boundary that carries one. The two cases were inverted:

  recordCount % watermarkBatchSize != 0 -> watermark 0, flag true
      -> re-persists a zero on 9999 of every 10000 writes
  recordCount % watermarkBatchSize == 0 -> watermark N, flag false
      -> drops the only value worth saving

The stored watermark therefore never left 0. Recovery stayed correct,
because replaying .idx from offset 0 is a superset of replaying from N
and replay is idempotent, so this never surfaced as a failure. It only
meant generateLevelDbFile walked the entire index on every rebuild, and
every needle write paid a second leveldb Put to rewrite the same zero.

Pass "watermark != 0" so the boundary write checkpoints and the writes
in between leave the key alone.

Verified on a 25000-needle volume: the stored watermark now reads 20000
instead of 0, and a rebuild replays 5000 entries instead of 25000.

The new test drives a full batch of Puts and a full batch of Deletes to
cover both call sites.

weed/storage/needle_map_leveldb.go

@@ -186,7 +186,7 @@ func (m *LevelDbNeedleMap) Put(key NeedleId, offset Offset, size Size) error {
 		watermark = (m.recordCount / watermarkBatchSize) * watermarkBatchSize
 		glog.V(1).Infof("put cnt:%d for %s,watermark: %d", m.recordCount, m.dbFileName, watermark)
 	}
-	return levelDbWrite(m.db, key, offset, size, watermark == 0, watermark)
+	return levelDbWrite(m.db, key, offset, size, watermark != 0, watermark)
 }
 
 func getWatermark(db *leveldb.DB) uint64 {
@@ -252,7 +252,7 @@ func (m *LevelDbNeedleMap) Delete(key NeedleId, offset Offset) error {
 	} else {
 		watermark = (m.recordCount / watermarkBatchSize) * watermarkBatchSize
 	}
-	return levelDbWrite(m.db, key, oldNeedle.Offset, -oldNeedle.Size, watermark == 0, watermark)
+	return levelDbWrite(m.db, key, oldNeedle.Offset, -oldNeedle.Size, watermark != 0, watermark)
 }
 
 func (m *LevelDbNeedleMap) Close() {

weed/storage/needle_map_leveldb_test.go

@@ -71,6 +71,43 @@ func TestGenerateLevelDbFileStaleWatermarkRebuilds(t *testing.T) {
 	}
 }
 
+// The write on a watermarkBatchSize boundary is the one that must persist the
+// replay watermark; the ordinary writes in between must leave it alone. Passing
+// "watermark == 0" inverted that and pinned the stored watermark at 0, so every
+// rebuild replayed the whole .idx. Deletes count too: tombstones are .idx
+// entries and advance the watermark the same way.
+func TestLevelDbNeedleMapWatermarkAdvances(t *testing.T) {
+	dir := t.TempDir()
+
+	indexFile, err := os.Create(filepath.Join(dir, "wm.idx"))
+	if err != nil {
+		t.Fatalf("create index file: %v", err)
+	}
+	m, err := NewLevelDbNeedleMap(filepath.Join(dir, "wm.ldb"), indexFile, nil, 0, needle.GetCurrentVersion())
+	if err != nil {
+		t.Fatalf("NewLevelDbNeedleMap: %v", err)
+	}
+	defer m.Close()
+
+	for i := uint64(1); i <= watermarkBatchSize; i++ {
+		if err := m.Put(types.Uint64ToNeedleId(i), types.ToOffset(int64(i*1024)), types.Size(512)); err != nil {
+			t.Fatalf("put %d: %v", i, err)
+		}
+	}
+	if got := getWatermark(m.db); got != watermarkBatchSize {
+		t.Errorf("after %d puts: watermark = %d, want %d", watermarkBatchSize, got, watermarkBatchSize)
+	}
+
+	for i := uint64(1); i <= watermarkBatchSize; i++ {
+		if err := m.Delete(types.Uint64ToNeedleId(i), types.ToOffset(int64(i*1024))); err != nil {
+			t.Fatalf("delete %d: %v", i, err)
+		}
+	}
+	if got, want := getWatermark(m.db), uint64(2*watermarkBatchSize); got != want {
+		t.Errorf("after %d deletes: watermark = %d, want %d", watermarkBatchSize, got, want)
+	}
+}
+
 func TestLevelDbNeedleMap_Concurrency(t *testing.T) {
 	dir, err := os.MkdirTemp("", "test_leveldb_concurrency")
 	if err != nil {