kvstorage/wag: prepare for WAG truncation

This commit sets up the premitives for WAG truncations by:

1) Adding a `Delete` function for removing WAG nodes by index.
2) Changing the WAG `Iterator.Iter` method to return
 `iter.Seq2[uint64, wagpb.Node]` pair. This will be useful when we
  iterate over the WAG and truncate the nodes that have been applied and
  synced.

Epic: none
Release note: None

Co-Authored-By: roachdev-claude <roachdev-claude-bot@cockroachlabs.com>

Author: iskettaneh

pkg/kv/kvserver/kvstorage/wag/BUILD.bazel

@@ -4,29 +4,41 @@ go_library(
     name = "wag",
     srcs = [
         "store.go",
+        "truncator.go",
         "writer.go",
     ],
     importpath = "github.com/cockroachdb/cockroach/pkg/kv/kvserver/kvstorage/wag",
     visibility = ["//visibility:public"],
     deps = [
         "//pkg/keys",
+        "//pkg/kv/kvpb",
+        "//pkg/kv/kvserver/kvserverpb",
         "//pkg/kv/kvserver/kvstorage/wag/wagpb",
+        "//pkg/roachpb",
         "//pkg/storage",
+        "//pkg/util/hlc",
+        "//pkg/util/log",
     ],
 )
 
 go_test(
     name = "wag_test",
-    srcs = ["store_test.go"],
+    srcs = [
+        "store_test.go",
+        "truncator_test.go",
+    ],
     data = glob(["testdata/**"]),
     embed = [":wag"],
     deps = [
+        "//pkg/keys",
         "//pkg/kv/kvpb",
+        "//pkg/kv/kvserver/kvserverpb",
         "//pkg/kv/kvserver/kvstorage/wag/wagpb",
         "//pkg/kv/kvserver/print",
         "//pkg/roachpb",
         "//pkg/storage",
         "//pkg/testutils/echotest",
+        "//pkg/util/hlc",
         "//pkg/util/leaktest",
         "//pkg/util/log",
         "@com_github_stretchr_testify//require",

pkg/kv/kvserver/kvstorage/wag/store.go

@@ -48,11 +48,16 @@ func Write(w storage.Writer, index uint64, node wagpb.Node) error {
 	return w.PutUnversioned(keys.StoreWAGNodeKey(index), data)
 }
 
+// Delete removes the WAG node at the given sequence number.
+func Delete(w storage.Writer, index uint64) error {
+	return w.ClearUnversioned(keys.StoreWAGNodeKey(index), storage.ClearOptions{})
+}
+
 // Iterator helps to scan the WAG sequence.
 //
 //	var iter wag.Iterator
-//	for node := range iter.Iter(ctx, reader) {
-//		// process node
+//	for index, node := range iter.Iter(ctx, reader) {
+//		// process index, node
 //	}
 //	if err := iter.Error(); err != nil {
 //		return err
@@ -64,8 +69,9 @@ type Iterator struct {
 	err error
 }
 
-// Iter returns an iterator that scans the WAG sequence.
-func (it *Iterator) Iter(ctx context.Context, r storage.Reader) iter.Seq[wagpb.Node] {
+// Iter returns an iterator that scans the WAG sequence. The iterator yields a
+// pair containing the WAG node index and the WAG node itself.
+func (it *Iterator) Iter(ctx context.Context, r storage.Reader) iter.Seq2[uint64, wagpb.Node] {
 	prefix := keys.StoreWAGPrefix()
 	mi, err := r.NewMVCCIterator(ctx, storage.MVCCKeyIterKind, storage.IterOptions{
 		UpperBound: prefix.PrefixEnd(),
@@ -76,13 +82,18 @@ func (it *Iterator) Iter(ctx context.Context, r storage.Reader) iter.Seq[wagpb.N
 	}
 	mi.SeekGE(storage.MakeMVCCMetadataKey(prefix))
 
-	return func(yield func(wagpb.Node) bool) {
+	return func(yield func(uint64, wagpb.Node) bool) {
 		defer mi.Close()
 		for ; ; mi.Next() {
 			if ok, err := mi.Valid(); err != nil || !ok {
 				it.err = err
 				return
 			}
+			index, err := keys.DecodeWAGNodeKey(mi.UnsafeKey().Key)
+			if err != nil {
+				it.err = err
+				return
+			}
 			v, err := mi.UnsafeValue()
 			if err != nil {
 				it.err = err
@@ -92,7 +103,7 @@ func (it *Iterator) Iter(ctx context.Context, r storage.Reader) iter.Seq[wagpb.N
 			if it.err = node.Unmarshal(v); it.err != nil { // nolint:protounmarshal
 				return
 			}
-			if !yield(node) {
+			if !yield(index, node) {
 				return
 			}
 		}