Subversion Repositories zfs_utils

# Simplified BSD License (FreeBSD License)

#

# Copyright (c) 2025, Daily Data Inc.

# All rights reserved.

#

# Redistribution and use in source and binary forms, with or without

# modification, are permitted provided that the following conditions are met:

#

# 1. Redistributions of source code must retain the above copyright notice, this

#    list of conditions and the following disclaimer.

#

# 2. Redistributions in binary form must reproduce the above copyright notice,

#    this list of conditions and the following disclaimer in the documentation

#    and/or other materials provided with the distribution.

#

# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"

# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE

# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE

# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE

# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL

# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR

# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER

# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,

# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE

# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

package ZFS_Utils;

use strict;

use warnings;

use Exporter 'import';

use Data::Dumper;

use POSIX qw(strftime);

use File::Path qw(make_path);

# library of ZFS related utility functions

# Copyright 2025 Daily Data Inc. <rodo@dailydata.net>

# currently used for sneakernet scripts, but plans to expand to other ZFS related tasks

# functions include:

#   runCmd: execute a command and return its output (captures exit status in $lastRunError;

#           supports optional stderr merge via $merge_stderr)

#   shredFile: securely delete a file using gshred (note: not effective on ZFS due to COW)

#   logMsg: timestamped logging to a file and optionally to console; respects $verboseLoggingLevel

#   loadConfig: load a YAML configuration file into a hashref; will create the file from a

#           provided default hashref if the file does not exist (uses YAML::XS or YAML::Tiny)

#   mountDriveByLabel: find and mount a drive by its GPT label (supports ufs/msdos; waits

#           for device and creates mountpoint)

#   unmountDriveByLabel: unmount a drive found by GPT label and remove the mountpoint if empty

#   mountGeli: high level orchestrator to decrypt multiple GELI devices and import/mount a ZFS pool

#   decryptAndMountGeli: attach GELI devices, optionally build a combined key, import the pool

#           and mount ZFS datasets

#   makeGeliKey: create a GELI key by XOR'ing a remote binary keyfile and a local 256-bit hex key;

#           writes a 32-byte binary key file with mode 0600

#   findGeliDisks: discover candidate disks suitable for GELI on the host

#   makeReplicateCommands: build zfs send/receive command lists from snapshot lists and prior status;

#           intelligently determines recursive vs per-filesystem sends and incremental vs full sends

#           based on snapshot availability; filters snapshots by matching parent path + dataset name

#           to avoid false matches with similarly-named datasets in different locations

#   sendReport: helper to deliver replication reports (email/file) — exported for scripts to implement

#   fatalError: helper to log a fatal condition and die (convenience wrapper)

#   getDirectoryList: utility to list directory contents with optional filters

#   cleanDirectory: safe directory cleaning utility used by snapshot pruning helpers

#   exported package variables: $logFileName, $displayLogsOnConsole, $lastRunError, $verboseLoggingLevel

#

# v1.0 RWR 20251215

# This is the initial, tested release

#

# v1.0.1 RWR 20251215

# Added verbose logging control to logMsg calls, controlled by $verboseLoggingLevel

#

# v1.1.0 RWR 20251217

# Added added version variable $VERSION

# Added more logging of source/target snapshots in doSourceReplication for debugging.

# max verbosity is now 5 instead of 3.

# optimized makeReplicateCommands to avoid unnecessary copies of large arrays and increase reliability

# when processing snapshots with inconsistent naming in child datasets.

# Exported functions and variables

our @EXPORT_OK = qw(loadConfig shredFile mountDriveByLabel unmountDriveByLabel mountGeli logMsg runCmd makeReplicateCommands sendReport fatalError getDirectoryList cleanDirectory $logFileName $displayLogsOnConsole $lastRunError $verboseLoggingLevel);

our $VERSION = '1.1.0';

# these are variables which affect the flow of the program and are exported so they can be modified by the caller

our $logFileName = '/tmp/zfs_utils.log'; # this can be overridden by the caller, and turned off with empty string

our $displayLogsOnConsole = 1; # if non-zero, log messages are also printed to console

our $merge_stderr = 0; # if set to 1, stderr is captured in runCmd

our $lastRunError = 0; # tracks the last error code from runCmd

our $verboseLoggingLevel = 0; # if non-zero, logMsg will include more verbose output

# Execute a command and return its output.

# If called in scalar context, returns the full output as a single string.

# If called in list context, returns the output split into lines.

# If $merge_stderr is true (default), stderr is merged into stdout (only for scalar commands).

# returns undef on failure and logs failure message.

sub runCmd {

   my $cmd = join( ' ', @_ );

   $merge_stderr = 1 unless defined $merge_stderr;

   my $output = '';

   logMsg( "Running command [$cmd]" ) if $verboseLoggingLevel >= 2;

   $cmd .= ' 2>&1' if $merge_stderr;

   $output = `$cmd`;

   $lastRunError = $?;

   if ( $lastRunError ) {

      if ($? == -1) {

         logMsg( "failed to execute: $!");

         return '';

      } elsif ($? & 127) { # fatal error, exit program

         logMsg( sprintf( "child died with signal %d, %s coredump\n", ($? & 127),  ($? & 128) ? 'with' : 'without' ) );

         die;

      } elsif ($? >> 8) { # it had some return code other than 0

         logMsg( sprintf( "child exited with value %d\n", $? >> 8 ) );

      }

   }

   $output //= '';

   if (wantarray) {

      return $output eq '' ? () : split(/\n/, $output);

   } else {

      return $output;

   }

}

# this calls gshred which will overwrite the file 3 times, then

# remove it.

# NOTE: this will not work on ZFS, since ZFS is CopyOnWrite (COW)

# so assuming file is on something without COW (ramdisk, UFS, etc)

sub shredFile {

   my $filename = shift;

   `/usr/local/bin/gshred -u -f -s 32 $filename` if -e $filename;

}

sub logMsg {

    my $msg = shift;

    my $filename = shift // $logFileName;

    my $timeStampFormat = shift // '%Y-%m-%d %H:%M:%S';

    my $timestamp = strftime($timeStampFormat, localtime());

    if (defined $filename && $filename ne '' ) {

       open my $logfh, '>>', $filename or die "Could not open log file $filename: $!\n";

       print $logfh "$timestamp\t$msg\n";

       close $logfh;

    }

    print "$timestamp\t$msg\n" if ($displayLogsOnConsole);

}

# find a drive by it's label by scanning /dev/gpt/

# driveInfo is a hashref with the following keys:

# label - the GPT label of the drive (required)

# filesystem - the filesystem type (default: ufs)

# mountPath - where to mount the drive (default: /mnt/label)

# timeout - how long to wait for the drive (default: 600 seconds)

# check_interval - how often to check for the drive (default: 15 seconds)

# If the drive is found, mount it on mountPath and return the mountPath.

# If not found, return empty string.

sub mountDriveByLabel {

   my ( $driveInfo ) = @_;

   unless ($driveInfo->{label}) {

      logMsg("mountDriveByLabel: No drive label provided");

      return '';

   }

   unless ( $driveInfo->{label} =~ /^[a-zA-Z0-9_\-]+$/ ) {

      logMsg("mountDriveByLabel: Invalid label '$driveInfo->{label}'");

      return '';

   }

   logMsg("mountDriveByLabel: Looking for drive with label '$driveInfo->{label}'") if $verboseLoggingLevel >= 1;

   # default to /mnt/label if not provided

   $driveInfo->{mountPath} //= "/mnt/$driveInfo->{label}"; # this is where we'll mount it if we find it

   $driveInfo->{fstype} //= 'ufs'; # default to mounting ufs

   # The location for the label depends on filesystem. Only providing access to ufs and msdos here for safety.

   # gpt labeled drives for ufs are in /dev/gpt/, for msdosfs in /dev/msdosfs/

   my $labelPath = $driveInfo->{fstype} eq 'msdos' ? "/dev/msdosfs/$driveInfo->{label}" : "/dev/gpt/$driveInfo->{label}"; 

   # drive already mounted, just return the path

   my $output = runCmd( "mount | grep '$driveInfo->{mountPath}'" );

   return $driveInfo->{mountPath} if ( $lastRunError == 0 ); # grep found it for us

   # default to 10 minutes (600 seconds) if not provided

   $driveInfo->{timeout} //= 600;

   # default to checking every minute if not provided

   $driveInfo->{check_interval} //= 15;

   # wait up to $timeout seconds for device to appear, checking every 10 seconds

   while ( $driveInfo->{timeout} > 0 ) {

      if ( -e "$labelPath" ) {

         last;

      } else {

         print "Waiting for drive labeled $driveInfo->{label}, looking in $labelPath\n";

         sleep $driveInfo->{check_interval};

         $driveInfo->{timeout} -= $driveInfo->{check_interval};

      }

    }

    # if we found it, mount and return mount path

    if ( -e "$labelPath" ) {

       # ensure mount point

       unless ( -d $driveInfo->{mountPath} || make_path($driveInfo->{mountPath}) ) {

         logMsg("Failed to create $driveInfo->{mountPath}: $!");

         return '';

       }

       # mount device

       runCmd( "mount -t $driveInfo->{fstype} $labelPath $driveInfo->{mountPath}" );

       if ( $lastRunError ) {

         logMsg("Failed to mount $labelPath on $driveInfo->{mountPath}: $!") if $verboseLoggingLevel >= 0;

         return '';

       }

       return $driveInfo->{mountPath};

    } else {

       return '';

    }

}

# finds and unmounts a drive defined by $driveInfo.

# on success, removes the mount point if empty.

sub unmountDriveByLabel {

   my ( $driveInfo ) = @_;

   unless ($driveInfo->{label}) {

      logMsg("unmountDriveByLabel: No drive label provided");

      return '';

   }

   unless ( $driveInfo->{label} =~ /^[a-zA-Z0-9_\-]+$/ ) {

      logMsg("unmountDriveByLabel: Invalid label '$driveInfo->{label}'");

      return '';

   }

   logMsg("unmountDriveByLabel: Looking for drive with label '$driveInfo->{label}'") if $verboseLoggingLevel >= 1;

   # default to /mnt/label if not provided

   $driveInfo->{mountPath} //= "/mnt/$driveInfo->{label}"; # this is where we'll mount it if we find it

   runCmd( "mount | grep '$driveInfo->{mountPath}'" );

   if ( $lastRunError ) {

     logMsg("Drive with label '$driveInfo->{label}' is not mounted") if $verboseLoggingLevel >= 2;

     return '';

   }

   # unmount device

   runCmd( "umount $driveInfo->{mountPath}" );

   if ( $lastRunError ) {

     logMsg("Failed to unmount $driveInfo->{mountPath}: $!");

     return '';

   }

   # and remove the directory if empty (find command will return empty string or one filename)

   rmdir $driveInfo->{mountPath} unless runCmd( "find $driveInfo->{mountPath} -mindepth 1 -print -quit");

   return $driveInfo->{mountPath};

}

## Load a YAML configuration file into a hashref.

## If the file does not exist, and a default hashref is provided,

## create the file by dumping the default to YAML, then return the default.

sub loadConfig {

    my ($filename, $default) = @_;

    # If no filename was provided, return default or empty hashref

    die "No filename provided to loadConfig\n" unless defined $filename;

    # If file doesn't exist but a default hashref was provided, try to

    # create the file by dumping the default to YAML, then return the default.

    unless (-e $filename) {

      logMsg("Config file $filename does not exist. Creating it with default values.");

      if ($default && ref $default eq 'HASH') {

         my $wrote = 0;

         eval {

               require YAML::XS;

               YAML::XS->import();

               YAML::XS::DumpFile($filename, $default);

               $wrote = 1;

               1;

         } or do {

               eval {

                  require YAML::Tiny;

                  YAML::Tiny->import();

                  my $yt = YAML::Tiny->new($default);

                  $yt->write($filename);

                  $wrote = 1;

                  1;

               } or do {

                  logMsg("No YAML writer available (YAML::XS or YAML::Tiny). Could not create $filename");

               };

         };

         die "Failed to write default config to $filename:$!\n" unless $wrote;

      } # if default

      # No default provided; nothing to create

      return {};

   } # unless -e $filename

   my $yaml;

   # Try YAML::XS first, fall back to YAML::Tiny

   eval {

      require YAML::XS;

      YAML::XS->import();

      $yaml = YAML::XS::LoadFile($filename);

      logMsg("using YAML::XS to load $filename") if $verboseLoggingLevel >= 3;

      1;

   } or do {

      eval {

         require YAML::Tiny;

         YAML::Tiny->import();

         $yaml = YAML::Tiny->read($filename);

         $yaml = $yaml->[0] if $yaml;  # YAML::Tiny returns an arrayref of documents

         logMsg("using YAML::Tiny to load $filename") if $verboseLoggingLevel >= 3;

         1;

      } or do {

         logMsg("No YAML parser installed (YAML::XS or YAML::Tiny). Skipping config load from $filename");

         return ($default && ref $default eq 'HASH') ? $default : {};

      };

   };

   # Ensure we have a hashref

   die "Config file $filename did not produce a HASH.\n" unless (defined $yaml && ref $yaml eq 'HASH');

   return $yaml;

}

## Mount a GELI-encrypted ZFS pool (high-level orchestration).

##

## Arguments:

##   $geliConfig - HASHREF containing GELI/ZFS mounting configuration. Expected keys include:

##       poolname        - name of the zpool to import

##       secureKey       - HASHREF with { label, keyfile, path } describing the keyfile disk

##       target          - path where the combined keyfile will be written

##       diskList        - OPTIONAL arrayref of disk device names (eg: ['ada0','ada1'])

##

## Behavior:

##   - Mounts the keyfile disk (using mountDriveByLabel), builds the combined key (makeGeliKey),

##     then calls decryptAndMountGeli to attach geli devices and import/mount the zpool.

##

## Returns:

##   Pool name (string) on success, empty string on error.

sub mountGeli {

   my $geliConfig = shift;

   logMsg( "geli config detected, attempting to mount geli disks" ) if $verboseLoggingLevel >= 0;

   # Can't continue at all if no pool name

   unless ( $geliConfig->{'poolname'} ) {

      logMsg "Could not find pool name in configuration file\n";

      return '';

   }

   # find the keyfile disk and mount it

   $geliConfig->{secureKey}->{path} = mountDriveByLabel( $geliConfig->{secureKey} );

   unless ( $geliConfig->{secureKey}->{path} ) {

      logMsg "Could not find or mount keyfile disk with label: " . $geliConfig->{secureKey}->{label};

      return '';

   }

   # create the combined geli keyfile in target location

   unless ( makeGeliKey( $geliConfig ) ) {

         logMsg "Could not create geli keyfile\n";

         return '';

      }

   # decrypt and mount the geli disks and zfs pool

   my $poolname = decryptAndMountGeli( $geliConfig );

   return $poolname;

}

## Discover disks suitable for GELI/ZFS use on the host.

##

## Returns an array of device names (eg: qw( ada0 ada1 )) that appear free for use.

## The routine collects all disks, excludes disks with existing partitions and those

## referenced by active zpools.

sub findGeliDisks {

   logMsg("Finding available disks for GELI/ZFS use") if $verboseLoggingLevel >= 2;

   # get all disks in system

   my %allDisks = map{ chomp $_ ; $_ => 1 } runCmd( "geom disk list | grep 'Geom name:' | rev | cut -d' ' -f1 | rev" );

   # get the disks with partitions

   my @temp = runCmd( "gpart show -p | grep '^=>'");  # -p prints just the disks without partitions

   # remove them from the list

   foreach my $disk ( @temp ) {

      $allDisks{$1} = 0 if ( $disk =~ m/^=>[\t\s0-9]+([a-z][a-z0-9]+)/ ) ;

   }

   # get disk which are currently used for zpools

   @temp = runCmd( "zpool status -LP | grep '/dev/'" );

   foreach my $disk ( @temp ) {

      $allDisks{$1} = 0 if  $disk =~ m|/dev/([a-z]+\d+)|;

   }

   # return only the disks which are free (value 1)

   return grep{ $allDisks{$_} == 1 } keys %allDisks;

}

## Decrypt GELI-encrypted disks and import/mount the ZFS pool.

##

## Arguments:

##   $geliConfig - HASHREF expected to contain:

##       poolname - zpool name to import

##       target   - path to the combined GELI keyfile created by makeGeliKey

##       diskList - OPTIONAL arrayref of disk device names (if omitted, findGeliDisks() is used)

##

## Behavior:

##   - Ensures the pool is not already imported

##   - Attaches (geli attach) each supplied disk using the keyfile

##   - Attempts to import the specified pool and runs `zfs mount -a` to mount datasets

##

## Returns:

##   Pool name (string) on success; empty string on failure.

sub decryptAndMountGeli {

   my ($geliConfig) = shift;

   # if no list of disks provided, try to find them

   $geliConfig->{'diskList'} //= [ findGeliDisks() ];

   my $diskList = $geliConfig->{'diskList'};

   my $poolname = $geliConfig->{'poolname'};

   my $keyfile = $geliConfig->{'target'};

   # check if the pool already attached (grep returns 0 on found, something else on not)

   runCmd( "zpool list -H -o name | grep $poolname" );

   return $poolname unless $lastRunError;

   unless ( -e $keyfile ) {

      logMsg "GELI keyfile $keyfile does not exist\n";

      return '';

   }

   my @decrypted_devices;

   # Decrypt each disk in the list

   foreach my $disk (@{$geliConfig->{'diskList'}}) {

      $disk = '/dev/' . $disk unless $disk =~ m|/dev|;

      unless ( -e $disk ) {

         logMsg "Disk $disk does not exist\n";

         return '';

      }

      # Derive the decrypted device name (.eli suffix on FreeBSD)

      my $decrypted = $disk . '.eli';

      # Decrypt using geli attach with the keyfile

      logMsg("Decrypting $disk with keyfile $keyfile") if $verboseLoggingLevel >= 2;

      runCmd("geli attach -p -k $geliConfig->{target} $disk");

      if ( $lastRunError) {

         logMsg "Failed to decrypt $disk (exit $lastRunError)\n" if $verboseLoggingLevel >= 3;

         next; # ignore failed disks and continue to see if we can import the pool

      }

      unless ( -e $decrypted ) {

         logMsg "Decrypted device $decrypted does not exist after geli attach\n" if $verboseLoggingLevel >= 0;

         return '';

      }

      push @decrypted_devices, $decrypted;

   }

   # Import the ZFS pool

   logMsg("Importing ZFS pool $poolname") if $verboseLoggingLevel >= 0;

   my @import_cmd = ('zpool', 'import');

   push @import_cmd, $poolname;

   runCmd("zpool import $poolname" );

   unless ( $lastRunError == 0 ) {

      logMsg("Failed to import zfs pool $poolname (exit $lastRunError)\n");

      return '';

   }

   # Mount the ZFS pool (zfs mount -a mounts all filesystems in the pool)

   logMsg("Mounting ZFS pool $poolname") if $verboseLoggingLevel >= 1;

   runCmd('zfs mount -a');

   unless ( $lastRunError == 0 ) {

      logMsg("Failed to mount zfs pool $poolname (exit $lastRunError)\n");

      return '';

   }

   logMsg("Successfully decrypted and mounted pool $poolname") if $verboseLoggingLevel >= 2;

   return $poolname;

}

## Create a GELI key by XOR'ing a remote binary keyfile and a local key (hex string).

##

## Expected input (via $geliConfig HASHREF):

##   $geliConfig->{secureKey}->{path} - directory where the remote keyfile resides

##   $geliConfig->{secureKey}->{keyfile} - filename of the remote 32-byte binary key

##   $geliConfig->{localKey} - 64-hex char string OR path to a file containing the hex

##   $geliConfig->{target} - path to write the resulting 32-byte binary key

##

## Behavior:

##   - Reads 32 bytes from the remote binary key

##   - Reads/cleans the 64-hex local key and converts it to 32 bytes

##   - XORs the two 32-byte buffers and writes the 32-byte result to $target with mode 0600

##

## Returns: 1 on success. Dies on unrecoverable errors.

sub makeGeliKey {

   my ( $geliConfig ) = @_;

   $geliConfig->{secureKey}->{keyfile} //= '';

   $geliConfig->{localKey} //= '';

   $geliConfig->{target} //= '';

   if ( $geliConfig->{target} && -f $geliConfig->{target} ) {

      logMsg "GELI target keyfile $geliConfig->{target} already exists. Not overwriting.\n" if $verboseLoggingLevel >= 2;

      return 1;

   }

   my $remote_keyfile = "$geliConfig->{secureKey}->{path}/$geliConfig->{secureKey}->{keyfile}";

   my $localKeyHexOrPath = $geliConfig->{localKey};

   my $target = $geliConfig->{target};

   if ( $geliConfig->{secureKey}->{keyfile} && $geliConfig->{localKey} ) {

      # we have what we need to proceed

      if ( -f $remote_keyfile ) {

         logMsg "Creating GELI keyfile at $geliConfig->{target} using remote keyfile " . $geliConfig->{secureKey}->{keyfile} . " and local key\n" 

            if $verboseLoggingLevel >= 2;

      } else {

         die "Remote keyfile " . $geliConfig->{secureKey}->{keyfile} . " does not exist\n";

      }

   }

   # Read remote binary key

   open my $rh, '<:raw', $remote_keyfile or die "Unable to open $remote_keyfile: $!\n";

   my $rbuf;

   my $read = read($rh, $rbuf, 32);

   close $rh;

   die "Failed to read 32 bytes from $remote_keyfile (got $read)\n" unless defined $read && $read == 32;

   # Get local hex string (either direct string or file contents)

   my $hex;

   if (-e $localKeyHexOrPath) {

      open my $lh, '<', $localKeyHexOrPath or die "Unable to open local key file $localKeyHexOrPath: $!\n";

      local $/ = undef;

      $hex = <$lh>;

      close $lh;

   } else {

      $hex = $localKeyHexOrPath;

   }

   # clean hex (remove whitespace/newlines and optional 0x)

   $hex =~ s/0x//g;

   $hex =~ s/[^0-9a-fA-F]//g;

   die "Local key must be 64 hex characters (256-bit)\n" unless length($hex) == 64;

   my $lbuf = pack('H*', $hex);

   die "Local key decoded to unexpected length " . length($lbuf) . "\n" unless length($lbuf) == 32;

   # XOR the two buffers

   my $out = '';

   for my $i (0 .. 31) {

      $out .= chr( ord(substr($rbuf, $i, 1)) ^ ord(substr($lbuf, $i, 1)) );

   }

   # Ensure target directory exists

   my ($vol, $dirs, $file) = ($target =~ m{^(/?)(.*/)?([^/]+)$});

   if ($dirs) {

      my $dir = $dirs;

      $dir =~ s{/$}{};

      unless (-d $dir) {

         require File::Path;

         File::Path::make_path($dir) or die "Failed to create directory $dir: $!\n";

      }

   }

   # Write out binary key and protect permissions

   open my $oh, '>:raw', $target or die "Unable to open $target for writing: $!\n";

   print $oh $out or die "Failed to write to $target: $!\n";

   close $oh;

   chmod 0600, $target;

   return 1;

}

# make a bunch of replicate commands and return them to the caller as a list

# $sourceSnapsRef - list of snapshots on source machine

# $targetSnapsRef - list of snapshots on target machine

# $dataset - The name of the dataset we are working on (same on both source and target)

# $sourceParent - The parent dataset of $dataset on source

# $targetParent - The parent dataset of $dataset on target

# $newStatusRef - A place to put the updated $targetSnapsRef

# returns hashref of commands to execute, of form

#    {$dataset} = "zfs send command"

# where $dataset above can be a child of $dataset

sub makeReplicateCommands {

   my ( $sourceSnapsRef, $targetSnapsRef, $dataset, $sourceParent, $targetParent, $newStatusRef ) = @_;

   # Ensure all array refs are defined (use empty arrays if not provided)

   $sourceSnapsRef ||= [];

   $targetSnapsRef     ||= [];

   $newStatusRef  ||= [];

   # Normalize parent paths: ensure they end with '/' unless empty

   # This makes path construction consistent later (e.g., "pool/" + "dataset")

   $sourceParent //= '';

   $sourceParent .= '/' unless $sourceParent eq '' or substr($sourceParent, -1) eq '/';

   $targetParent //= '';

   $targetParent .= '/' unless $targetParent eq '' or substr($targetParent, -1) eq '/';

   logMsg( "makeReplicateCommands: dataset=[$dataset] sourceParent=[$sourceParent] targetParent=[$targetParent]" ) if $verboseLoggingLevel >= 4;

   logMsg( "makeReplicateCommands: source snapshots count=" . scalar(@$sourceSnapsRef) . ", target snapshots count=" . scalar(@$targetSnapsRef) ) if $verboseLoggingLevel >= 4;

   if ($verboseLoggingLevel >= 5) {

      logMsg( "makeReplicateCommands: RAW target snapshots BEFORE filtering:" );

      foreach my $snap (@$targetSnapsRef) {

         logMsg( "  [$snap]" );

      }

      logMsg( "makeReplicateCommands: RAW source snapshots BEFORE filtering:" );

      foreach my $snap (@$sourceSnapsRef) {

         logMsg( "  [$snap]" );

      }

   }

   my %commands; # Hash to store generated zfs send commands, keyed by filesystem name

   fatalError( "No dataset defined in makeReplicateCommands, can not continue") unless $dataset;

   # Filter snapshot lists to only include snapshots matching our dataset and its children

   # The dataset should match as a full path component (not substring)

   # Then strip the parent path prefix from each snapshot name

   # Example: "storage/mydata@snap1" becomes "mydata@snap1" when sourceParent="storage/"

   # This allows us to work with relative paths and handle different parent paths on source/target

   # Match: storage/mydata@snap, storage/mydata/child@snap

   # Don't match: storage/mydataset@snap (if dataset is "mydata")

   # Don't match: storage/otherparent/mydata@snap (different parent path)

   my $targetSnaps = [ map{ s/^$targetParent//r } grep{ /^\Q$targetParent$dataset\E(?:\/|@)/ } @$targetSnapsRef ];

   my $sourceSnaps = [ map{ s/^$sourceParent//r } grep{ /^\Q$sourceParent$dataset\E(?:\/|@)/ } @$sourceSnapsRef ];

   logMsg( "makeReplicateCommands: filtered source snapshots count=" . scalar(@$sourceSnaps) . ", filtered target snapshots count=" . scalar(@$targetSnaps) ) if $verboseLoggingLevel >= 4;

   logMsg( "makeReplicateCommands: filtered source snapshots: " . join(', ', sort @$sourceSnaps) ) if $verboseLoggingLevel >= 5;

   logMsg( "makeReplicateCommands: filtered target snapshots: " . join(', ', sort @$targetSnaps) ) if $verboseLoggingLevel >= 5;

   # Parse source snapshots to build a hash indexed by filesystem

   # Input lines may have format: "pool/fs@snapshot extra data"

   # We extract just the first token (pool/fs@snapshot) and split it into filesystem and snapshot name

   # Result: %snaps_by_fs = { "pool/fs" => ["snap1", "snap2", ...] }

   # This groups all snapshots by their parent filesystem

   my %snaps_by_fs;

   foreach my $line (@$sourceSnaps) {

      next unless defined $line && $line =~ /\S/;  # Skip empty lines

      my ($tok) = split /\s+/, $line;              # Get first token

      next unless $tok && $tok =~ /@/;             # Must contain @ separator

      my ($fs, $snap) = split /@/, $tok, 2;        # Split into filesystem and snapshot name

      push @{ $snaps_by_fs{$fs} }, $snap;          # Add snapshot to this filesystem's list

   }

   logMsg( "makeReplicateCommands: parsed filesystems: " . join(', ', sort keys %snaps_by_fs) ) if $verboseLoggingLevel >= 4;

   # If no snapshots were found, return empty array (nothing to replicate)

   return [] unless keys %snaps_by_fs;

   # Determine the root filesystem for recursive operations

   # We try to get it from the first non-empty snapshot line, otherwise use first sorted key

   # The root filesystem is used when we can do a single recursive send instead of multiple sends

   my ($first_line) = grep { defined $_ && $_ =~ /\S/ } @$sourceSnaps;

   my ($root_fs) = $first_line ? (split(/\s+/, $first_line))[0] =~ /@/ ? (split(/@/, (split(/\s+/, $first_line))[0]))[0] : undef : undef;

   $root_fs ||= (sort keys %snaps_by_fs)[0];

   # Build a hash of the most recent snapshot on target for each filesystem

   # This tells us what's already been replicated, so we can do incremental sends

   # If a filesystem isn't in this hash, we need to do a full (non-incremental) send

   # Note: If multiple snapshots exist for a filesystem, we keep only the last one

   # (later entries override earlier ones in the hash assignment)

   my %last_status_for;

   for my $s (@$targetSnaps) {

      next unless $s && $s =~ /@/;

      my ($fs, $snap) = split /@/, $s, 2;

      $last_status_for{$fs} = $snap;    # later entries override earlier ones -> last occurrence kept

   }

   if ($verboseLoggingLevel >= 4) {

      logMsg( "makeReplicateCommands: last status snapshots:" );

      for my $fs (sort keys %last_status_for) {

         logMsg( "  $fs => $last_status_for{$fs}" );

      }

   }

   # Build "from" and "to" snapshot mappings for each filesystem

   # "to" = the newest snapshot on source (what we want to send)

   # "from" = the last replicated snapshot on target (what we're sending from)

   # If "from" is undef, this filesystem hasn't been replicated before -> full send needed

   # Example: from="daily-2025-12-15" to="daily-2025-12-17" -> incremental send

   #          from=undef to="daily-2025-12-17" -> full send

   # NOTE: The "from" snapshot must exist in the source's snapshot list for incremental send

   #       If it doesn't exist in source, we need to find a common snapshot or do full send

   my %from_for;

   my %to_for;

   foreach my $fs (keys %snaps_by_fs) {

      my $arr = $snaps_by_fs{$fs};              # Get all snapshots for this filesystem

      next unless @$arr;                        # Skip if no snapshots

      $to_for{$fs} = $arr->[-1];                # Last element = newest snapshot to send

      # Check if the target's last status snapshot exists in the source list

      # If it does, we can do incremental send from that point

      # If it doesn't, the target may have a snapshot the source doesn't have anymore

      my $target_last = $last_status_for{$fs};

      if (defined $target_last && grep { $_ eq $target_last } @$arr) {

         $from_for{$fs} = $target_last;         # Use target's last snapshot as "from"

      } else {

         # Target's snapshot doesn't exist in source list - need full send

         $from_for{$fs} = undef;

      }

   }

   if ($verboseLoggingLevel >= 4) {

      logMsg( "makeReplicateCommands: from/to mapping:" );

      for my $fs (sort keys %to_for) {

         my $from = $from_for{$fs} // '(none - full send)';

         my $send_type = defined $from_for{$fs} ? 'incremental' : 'full';

         logMsg( "  $fs: from=$from to=$to_for{$fs} [$send_type]" );

      }

   }

   # Optimization check: Can we do a single recursive send?

   # Recursive sends are more efficient when replicating entire filesystem hierarchies

   # Condition: all filesystems must be sending to the same-named snapshot

   # Example: If pool/data@daily-2025-12-17 and pool/data/child@daily-2025-12-17 exist,

   #          we can do "zfs send -R pool/data@daily-2025-12-17" instead of two separate sends

   my %to_names = map { $_ => 1 } values %to_for;  # Get unique "to" snapshot names

   my $single_to_name = (keys %to_names == 1) ? (keys %to_names)[0] : undef;

   logMsg( "makeReplicateCommands: single_to_name=" . ($single_to_name // '(none - varied snapshots)') ) if $verboseLoggingLevel >= 4;

   if ($single_to_name) {

      # All filesystems are targeting the same snapshot name