Fix accounting for dual channel RDB bytes in replication stats (#2602)

Resolves #2545 

Followed the steps to reproduce the issue, and was able to get non-zero
`total_net_repl_output_bytes`.

```
(base) ~/workspace/valkey git:[fix-bug-2545]
src/valkey-cli INFO | grep total_net_repl_output_bytes
total_net_repl_output_bytes:1788
```

---------

Signed-off-by: Sarthak Aggarwal <[email protected]>

Author: sarthakaggarwal97

src/childinfo.c

@@ -36,6 +36,7 @@ typedef struct {
     size_t cow;
     monotime cow_updated;
     double progress;
+    size_t repl_output_bytes;
     childInfoType information_type; /* Type of information */
 } child_info_data;
 
@@ -64,7 +65,7 @@ void closeChildInfoPipe(void) {
 }
 
 /* Send save data to parent. */
-void sendChildInfoGeneric(childInfoType info_type, size_t keys, double progress, char *pname) {
+void sendChildInfoGeneric(childInfoType info_type, size_t keys, size_t repl_output_bytes, double progress, char *pname) {
     if (server.child_info_pipe[1] == -1) return;
 
     static monotime cow_updated = 0;
@@ -101,6 +102,7 @@ void sendChildInfoGeneric(childInfoType info_type, size_t keys, double progress,
 
     data.information_type = info_type;
     data.keys = keys;
+    data.repl_output_bytes = repl_output_bytes;
     data.cow = cow;
     data.cow_updated = cow_updated;
     data.progress = progress;
@@ -115,7 +117,7 @@ void sendChildInfoGeneric(childInfoType info_type, size_t keys, double progress,
 }
 
 /* Update Child info. */
-void updateChildInfo(childInfoType information_type, size_t cow, monotime cow_updated, size_t keys, double progress) {
+void updateChildInfo(childInfoType information_type, size_t cow, monotime cow_updated, size_t keys, size_t repl_output_bytes, double progress) {
     if (cow > server.stat_current_cow_peak) server.stat_current_cow_peak = cow;
 
     if (information_type == CHILD_INFO_TYPE_CURRENT_INFO) {
@@ -129,14 +131,16 @@ void updateChildInfo(childInfoType information_type, size_t cow, monotime cow_up
         server.stat_rdb_cow_bytes = server.stat_current_cow_peak;
     } else if (information_type == CHILD_INFO_TYPE_MODULE_COW_SIZE) {
         server.stat_module_cow_bytes = server.stat_current_cow_peak;
+    } else if (information_type == CHILD_INFO_TYPE_REPL_OUTPUT_BYTES) {
+        server.stat_net_repl_output_bytes += (long long)repl_output_bytes;
     }
 }
 
 /* Read child info data from the pipe.
  * if complete data read into the buffer,
  * data is stored into *buffer, and returns 1.
  * otherwise, the partial data is left in the buffer, waiting for the next read, and returns 0. */
-int readChildInfo(childInfoType *information_type, size_t *cow, monotime *cow_updated, size_t *keys, double *progress) {
+int readChildInfo(childInfoType *information_type, size_t *cow, monotime *cow_updated, size_t *keys, size_t *repl_output_bytes, double *progress) {
     /* We are using here a static buffer in combination with the server.child_info_nread to handle short reads */
     static child_info_data buffer;
     ssize_t wlen = sizeof(buffer);
@@ -156,6 +160,7 @@ int readChildInfo(childInfoType *information_type, size_t *cow, monotime *cow_up
         *cow = buffer.cow;
         *cow_updated = buffer.cow_updated;
         *keys = buffer.keys;
+        *repl_output_bytes = buffer.repl_output_bytes;
         *progress = buffer.progress;
         return 1;
     } else {
@@ -170,11 +175,12 @@ void receiveChildInfo(void) {
     size_t cow;
     monotime cow_updated;
     size_t keys;
+    size_t repl_output_bytes;
     double progress;
     childInfoType information_type;
 
     /* Drain the pipe and update child info so that we get the final message. */
-    while (readChildInfo(&information_type, &cow, &cow_updated, &keys, &progress)) {
-        updateChildInfo(information_type, cow, cow_updated, keys, progress);
+    while (readChildInfo(&information_type, &cow, &cow_updated, &keys, &repl_output_bytes, &progress)) {
+        updateChildInfo(information_type, cow, cow_updated, keys, repl_output_bytes, progress);
     }
 }

src/module.c

@@ -11480,7 +11480,7 @@ int VM_Fork(ValkeyModuleForkDoneHandler cb, void *user_data) {
  * reported in INFO.
  * The `progress` argument should between 0 and 1, or -1 when not available. */
 void VM_SendChildHeartbeat(double progress) {
-    sendChildInfoGeneric(CHILD_INFO_TYPE_CURRENT_INFO, 0, progress, "Module fork");
+    sendChildInfoGeneric(CHILD_INFO_TYPE_CURRENT_INFO, 0, 0, progress, "Module fork");
 }
 
 /* Call from the child process when you want to terminate it.

src/rdb.c

@@ -3723,6 +3723,9 @@ int saveSnapshotToConnectionSockets(rdbSnapshotOptions options) {
 
         if (retval == C_OK) {
             sendChildCowInfo(CHILD_INFO_TYPE_RDB_COW_SIZE, "RDB");
+            if (!options.use_pipe) {
+                sendChildInfoGeneric(CHILD_INFO_TYPE_REPL_OUTPUT_BYTES, 0, rdb.processed_bytes, -1, "RDB");
+            }
         }
         if (!options.use_pipe) {
             rioFreeConnset(&rdb);

src/server.c

@@ -6789,11 +6789,11 @@ int serverFork(int purpose) {
 }
 
 void sendChildCowInfo(childInfoType info_type, char *pname) {
-    sendChildInfoGeneric(info_type, 0, -1, pname);
+    sendChildInfoGeneric(info_type, 0, 0, -1, pname);
 }
 
 void sendChildInfo(childInfoType info_type, size_t keys, char *pname) {
-    sendChildInfoGeneric(info_type, keys, -1, pname);
+    sendChildInfoGeneric(info_type, keys, 0, -1, pname);
 }
 
 /* Dismiss big chunks of memory inside a client structure, see zmadvise_dontneed() */

src/server.h

@@ -1657,7 +1657,8 @@ typedef enum childInfoType {
     CHILD_INFO_TYPE_CURRENT_INFO,
     CHILD_INFO_TYPE_AOF_COW_SIZE,
     CHILD_INFO_TYPE_RDB_COW_SIZE,
-    CHILD_INFO_TYPE_MODULE_COW_SIZE
+    CHILD_INFO_TYPE_MODULE_COW_SIZE,
+    CHILD_INFO_TYPE_REPL_OUTPUT_BYTES
 } childInfoType;
 
 struct valkeyServer {
@@ -3156,7 +3157,7 @@ int rewriteSlotToAppendOnlyFileRio(rio *aof, int db_num, int hashslot, size_t *k
 /* Child info */
 void openChildInfoPipe(void);
 void closeChildInfoPipe(void);
-void sendChildInfoGeneric(childInfoType info_type, size_t keys, double progress, char *pname);
+void sendChildInfoGeneric(childInfoType info_type, size_t keys, size_t repl_output_bytes, double progress, char *pname);
 void sendChildCowInfo(childInfoType info_type, char *pname);
 void sendChildInfo(childInfoType info_type, size_t keys, char *pname);
 void receiveChildInfo(void);

tests/integration/dual-channel-replication.tcl

@@ -77,6 +77,48 @@ start_server {tags {"dual-channel-replication external:skip"}} {
                 fail "Replicas and primary offsets were unable to match."
             }
         }
+
+        test "Dual-channel replication counts snapshot bytes" {
+            wait_for_condition 50 100 {
+                [getInfoProperty [$primary info stats] "total_net_repl_output_bytes"] > 0
+            } else {
+                fail "Replication output bytes not updated"
+            }
+        }
+    }
+}
+
+start_server {tags {"dual-channel-replication external:skip"}} {
+    set replica [srv 0 client]
+    set replica_host [srv 0 host]
+    set replica_port [srv 0 port]
+    start_server {} {
+        set primary [srv 0 client]
+        set primary_host [srv 0 host]
+        set primary_port [srv 0 port]
+
+        $primary config set repl-diskless-sync yes
+        $primary config set repl-diskless-sync-delay 0
+        $primary config set dual-channel-replication-enabled yes
+
+        $replica config set repl-diskless-sync yes
+        $replica config set repl-diskless-load swapdb
+        $replica config set dual-channel-replication-enabled yes
+
+        for {set j 0} {$j < 100} {incr j} {
+            $primary set key$j [string repeat x 100]
+        }
+        $primary config resetstat
+
+        test "dual-channel replication reports rdb transfer bytes" {
+            $replica replicaof $primary_host $primary_port
+            verify_replica_online $primary 0 700
+            wait_for_condition 50 100 {
+                [getInfoProperty [$primary info stats] "total_net_repl_output_bytes"] > 1000
+            } else {
+                fail "Replication output bytes not updated"
+            }
+        }
     }
 }