Mojo/S3/ResumableMultipartUpload.pm

gistfile1.txt

package Mojo::S3::ResumableMultipartUpload;

use Mojo::Base -base, -signatures;

use Carp qw(croak);
use File::Basename qw(basename);
use File::Spec;
use File::Temp qw(tempfile);
use Mojo::File qw(path);
use Mojo::JSON qw(decode_json encode_json);
use Mojo::URL;
use Mojo::UserAgent;
use Mojo::Util qw(xml_escape);
use POSIX qw(ceil);
use Time::HiRes qw(sleep);

our $VERSION = '0.01';

has access_key    => sub { $ENV{AWS_ACCESS_KEY_ID}     // $ENV{AWS_ACCESS_KEY} };
has secret_key    => sub { $ENV{AWS_SECRET_ACCESS_KEY} // $ENV{AWS_SECRET_KEY} };
has session_token => sub { $ENV{AWS_SESSION_TOKEN} };
has region        => sub { $ENV{AWS_REGION} // $ENV{AWS_DEFAULT_REGION} // 'us-east-1' };

has bucket => undef;
has key    => undef;

# Set this for S3-compatible services, for example:
#   https://nyc3.digitaloceanspaces.com
#
# Amazon S3 defaults to virtual-hosted-style URLs.
has endpoint => undef;

# Custom endpoints commonly need path-style addressing.
has path_style => sub ($self) {
  return defined $self->endpoint ? 1 : 0;
};

has part_size   => 64 * 1024 * 1024;
has max_retries => 5;
has retry_delay => 1;
has debug       => 0;

has content_type => 'application/octet-stream';

has ua => sub {
  return Mojo::UserAgent
    ->with_roles('+AWSSignature4')
    ->new;
};

has on_progress => undef;

sub upload ($self, $filename, %options) {
  croak 'filename is required' unless defined $filename && length $filename;
  croak "File does not exist: $filename" unless -f $filename;
  croak "File is not readable: $filename" unless -r $filename;

  $self->_validate_configuration;

  my @stat = stat $filename;
  my $file_size = $stat[7];
  my $mtime     = $stat[9];

  croak 'Multipart upload cannot upload an empty file'
    unless $file_size > 0;

  my $part_size = $options{part_size} // $self->part_size;
  $part_size = $self->_normalize_part_size($file_size, $part_size);

  my $state_file = $options{state_file}
    // $self->_default_state_file($filename);

  my $state = $self->_load_or_create_state(
    filename   => $filename,
    state_file => $state_file,
    file_size  => $file_size,
    mtime      => $mtime,
    part_size  => $part_size,
  );

  # S3 is authoritative. Reconcile the local state with the parts that
  # currently exist for this upload ID.
  my $remote_parts = $self->list_parts($state->{upload_id});

  $state->{parts} = {
    map {
      $_ => {
        etag => $remote_parts->{$_}{etag},
        size => $remote_parts->{$_}{size},
      }
    } keys %$remote_parts
  };

  $self->_save_state($state_file, $state);

  my $part_count = ceil($file_size / $part_size);

  open my $source, '<:raw', $filename
    or croak "Cannot open $filename: $!";

  for my $part_number (1 .. $part_count) {
    my $offset = ($part_number - 1) * $part_size;
    my $length = $file_size - $offset;
    $length = $part_size if $length > $part_size;

    if ($self->_remote_part_matches(
      $state->{parts}{$part_number},
      $length,
    )) {
      $self->_report_progress(
        event       => 'skipped',
        part_number => $part_number,
        part_count  => $part_count,
        bytes       => $length,
        offset      => $offset,
        file_size   => $file_size,
        upload_id   => $state->{upload_id},
      );

      next;
    }

    my $part_file = $self->_extract_part(
      source      => $source,
      source_name => $filename,
      offset      => $offset,
      length      => $length,
      part_number => $part_number,
    );

    my $etag;

    my $ok = eval {
      $etag = $self->upload_part(
        upload_id   => $state->{upload_id},
        part_number => $part_number,
        filename    => $part_file,
      );

      1;
    };

    my $error = $@;
    unlink $part_file;

    die $error unless $ok;

    $state->{parts}{$part_number} = {
      etag => $etag,
      size => $length,
    };

    $self->_save_state($state_file, $state);

    $self->_report_progress(
      event       => 'uploaded',
      part_number => $part_number,
      part_count  => $part_count,
      bytes       => $length,
      offset      => $offset,
      file_size   => $file_size,
      upload_id   => $state->{upload_id},
    );
  }

  close $source
    or croak "Cannot close $filename: $!";

  my @parts = map {
    +{
      part_number => $_,
      etag        => $state->{parts}{$_}{etag},
    }
  } 1 .. $part_count;

  my $result = $self->complete_multipart_upload(
    upload_id => $state->{upload_id},
    parts     => \@parts,
  );

  unlink $state_file
    or warn "Could not remove completed upload state $state_file: $!"
      if -e $state_file;

  $self->_report_progress(
    event      => 'completed',
    part_count => $part_count,
    file_size  => $file_size,
    upload_id  => $state->{upload_id},
    etag       => $result->{etag},
    location   => $result->{location},
  );

  return {
    %$result,
    bucket     => $self->bucket,
    key        => $self->key,
    upload_id  => $state->{upload_id},
    file_size  => $file_size,
    part_count => $part_count,
  };
}

sub create_multipart_upload ($self) {
  my $url = $self->_object_url;
  $url->query->append(uploads => undef);

  my ($fh, $payload) = tempfile('s3-create-XXXXXXXX', TMPDIR => 1);
  binmode $fh;
  close $fh;

  my $tx = eval {
    $self->_request(
      method       => 'POST',
      url          => $url,
      content_file => $payload,
      headers      => {
        'Content-Type' => $self->content_type,
      },
    );
  };

  my $error = $@;
  unlink $payload;
  die $error if $error;

  my $dom = $tx->res->dom;

  my $upload_id = $dom->at('UploadId')?.all_text;
  croak 'CreateMultipartUpload response did not contain UploadId'
    unless defined $upload_id && length $upload_id;

  return $upload_id;
}

sub upload_part ($self, %args) {
  my $upload_id   = $args{upload_id}
    // croak 'upload_id is required';
  my $part_number = $args{part_number}
    // croak 'part_number is required';
  my $filename    = $args{filename}
    // croak 'filename is required';

  croak 'part_number must be between 1 and 10000'
    unless $part_number =~ /^\d+$/
    && $part_number >= 1
    && $part_number <= 10_000;

  my $url = $self->_object_url;
  $url->query(
    partNumber => $part_number,
    uploadId   => $upload_id,
  );

  my $tx = $self->_request(
    method       => 'PUT',
    url          => $url,
    content_file => $filename,
    headers      => {
      'Content-Type' => 'application/octet-stream',
    },
  );

  my $etag = $tx->res->headers->etag;

  croak "UploadPart $part_number response did not contain an ETag"
    unless defined $etag && length $etag;

  return $etag;
}

sub list_parts ($self, $upload_id) {
  croak 'upload_id is required'
    unless defined $upload_id && length $upload_id;

  my %parts;
  my $marker;

  while (1) {
    my $url = $self->_object_url;

    my @query = (
      uploadId => $upload_id,
      'max-parts' => 1000,
    );

    push @query, ('part-number-marker' => $marker)
      if defined $marker;

    $url->query(@query);

    my ($fh, $payload) = tempfile('s3-list-XXXXXXXX', TMPDIR => 1);
    binmode $fh;
    close $fh;

    my $tx = eval {
      $self->_request(
        method       => 'GET',
        url          => $url,
        content_file => $payload,
      );
    };

    my $error = $@;
    unlink $payload;
    die $error if $error;

    my $dom = $tx->res->dom;

    for my $part ($dom->find('Part')->each) {
      my $number = $part->at('PartNumber')?.all_text;
      my $etag   = $part->at('ETag')?.all_text;
      my $size   = $part->at('Size')?.all_text;

      next unless defined $number && defined $etag;

      $parts{$number} = {
        etag => $etag,
        size => 0 + ($size // 0),
      };
    }

    my $truncated = $dom->at('IsTruncated')?.all_text // 'false';
    last unless lc($truncated) eq 'true';

    $marker = $dom->at('NextPartNumberMarker')?.all_text;

    croak 'Truncated ListParts response omitted NextPartNumberMarker'
      unless defined $marker;
  }

  return \%parts;
}

sub complete_multipart_upload ($self, %args) {
  my $upload_id = $args{upload_id}
    // croak 'upload_id is required';
  my $parts = $args{parts}
    // croak 'parts is required';

  croak 'parts must be a non-empty array reference'
    unless ref $parts eq 'ARRAY' && @$parts;

  my $xml = '<CompleteMultipartUpload>';

  for my $part (sort {
    $a->{part_number} <=> $b->{part_number}
  } @$parts) {
    croak 'Every part requires part_number and etag'
      unless defined $part->{part_number}
      && defined $part->{etag};

    $xml .= '<Part>';
    $xml .= '<PartNumber>'
      . xml_escape($part->{part_number})
      . '</PartNumber>';
    $xml .= '<ETag>'
      . xml_escape($part->{etag})
      . '</ETag>';
    $xml .= '</Part>';
  }

  $xml .= '</CompleteMultipartUpload>';

  my ($fh, $payload) = tempfile('s3-complete-XXXXXXXX', TMPDIR => 1);
  binmode $fh;
  print {$fh} $xml
    or croak "Cannot write completion payload: $!";
  close $fh
    or croak "Cannot close completion payload: $!";

  my $url = $self->_object_url;
  $url->query(uploadId => $upload_id);

  my $tx = eval {
    $self->_request(
      method       => 'POST',
      url          => $url,
      content_file => $payload,
      headers      => {
        'Content-Type' => 'application/xml',
      },
    );
  };

  my $error = $@;
  unlink $payload;
  die $error if $error;

  # CompleteMultipartUpload can return HTTP 200 while embedding an Error
  # document in the response body.
  my $dom = $tx->res->dom;

  if (my $error_node = $dom->at('Error')) {
    my $code    = $error_node->at('Code')?.all_text    // 'Unknown';
    my $message = $error_node->at('Message')?.all_text // 'Unknown S3 error';

    croak "CompleteMultipartUpload failed: $code: $message";
  }

  return {
    location => $dom->at('Location')?.all_text,
    bucket   => $dom->at('Bucket')?.all_text,
    key      => $dom->at('Key')?.all_text,
    etag     => $dom->at('ETag')?.all_text,
  };
}

sub abort_multipart_upload ($self, $upload_id) {
  croak 'upload_id is required'
    unless defined $upload_id && length $upload_id;

  my $url = $self->_object_url;
  $url->query(uploadId => $upload_id);

  my ($fh, $payload) = tempfile('s3-abort-XXXXXXXX', TMPDIR => 1);
  binmode $fh;
  close $fh;

  my $tx = eval {
    $self->_request(
      method       => 'DELETE',
      url          => $url,
      content_file => $payload,
    );
  };

  my $error = $@;
  unlink $payload;
  die $error if $error;

  return 1;
}

sub _request ($self, %args) {
  my $method       = uc($args{method} // croak 'method is required');
  my $url          = $args{url}       // croak 'url is required';
  my $headers      = $args{headers}   // {};
  my $content_file = $args{content_file};

  croak 'content_file is required by the AWSSignature4 role'
    unless defined $content_file;

  my %signing = (
    service      => 's3',
    region       => $self->region,
    access_key   => $self->access_key,
    secret_key   => $self->secret_key,
    content      => $content_file,
    debug        => $self->debug,
  );

  $signing{session_token} = $self->session_token
    if defined $self->session_token
    && length $self->session_token;

  my $attempt = 0;
  my $last_error;

  while ($attempt <= $self->max_retries) {
    $attempt++;

    my $tx = eval {
      my $method_name = lc $method;

      $self->ua->$method_name(
        $url => $headers => awssig4 => \%signing
      )->result;
    };

    if (!$@ && $tx && $tx->res) {
      my $status = $tx->res->code // 0;

      return $tx if $status >= 200 && $status < 300;

      $last_error = $self->_s3_error($tx);

      # Do not repeatedly retry normal client errors, except the errors that
      # are commonly transient or caused by throttling.
      my $retryable = $status == 408
        || $status == 409
        || $status == 429
        || $status >= 500;

      croak $last_error unless $retryable;
    }
    else {
      $last_error = $@ || 'Request failed without a response';
    }

    last if $attempt > $self->max_retries;

    my $delay = $self->retry_delay * (2 ** ($attempt - 1));
    $delay += rand($delay * 0.25);

    warn sprintf(
      "S3 request attempt %d failed; retrying in %.2f seconds: %s\n",
      $attempt,
      $delay,
      $last_error,
    ) if $self->debug;

    sleep $delay;
  }

  croak $last_error // 'S3 request failed';
}

sub _s3_error ($self, $tx) {
  my $status  = $tx->res->code    // 0;
  my $reason  = $tx->res->message // 'Unknown HTTP error';
  my $request = $tx->res->headers->header('x-amz-request-id');

  my $body = $tx->res->body // '';
  my ($code, $message);

  if (length $body) {
    my $dom = eval { $tx->res->dom };

    if ($dom) {
      $code    = $dom->at('Code')?.all_text;
      $message = $dom->at('Message')?.all_text;
    }
  }

  my $error = "S3 request failed: HTTP $status $reason";
  $error .= "; $code"               if defined $code;
  $error .= ": $message"            if defined $message;
  $error .= "; request-id=$request" if defined $request;

  return $error;
}

sub _load_or_create_state ($self, %args) {
  my $state_file = $args{state_file};

  if (-e $state_file) {
    my $state = eval {
      decode_json(path($state_file)->slurp);
    };

    croak "Cannot parse upload state $state_file: $@"
      if $@ || ref $state ne 'HASH';

    $self->_validate_state($state, %args);
    return $state;
  }

  my $upload_id = $self->create_multipart_upload;

  my $state = {
    version     => 1,
    upload_id   => $upload_id,
    filename    => File::Spec->rel2abs($args{filename}),
    file_size   => 0 + $args{file_size},
    mtime       => 0 + $args{mtime},
    part_size   => 0 + $args{part_size},
    bucket      => $self->bucket,
    key         => $self->key,
    region      => $self->region,
    endpoint    => $self->endpoint,
    path_style  => $self->path_style ? 1 : 0,
    parts       => {},
  };

  $self->_save_state($state_file, $state);

  return $state;
}

sub _validate_state ($self, $state, %args) {
  my $absolute_filename = File::Spec->rel2abs($args{filename});

  my @checks = (
    ['filename',   $absolute_filename],
    ['file_size',  0 + $args{file_size}],
    ['mtime',      0 + $args{mtime}],
    ['part_size',  0 + $args{part_size}],
    ['bucket',     $self->bucket],
    ['key',        $self->key],
    ['region',     $self->region],
    ['path_style', $self->path_style ? 1 : 0],
  );

  for my $check (@checks) {
    my ($name, $expected) = @$check;
    my $actual = $state->{$name};

    croak "Upload state mismatch for $name"
      unless defined $actual
      && defined $expected
      && "$actual" eq "$expected";
  }

  my $expected_endpoint = $self->endpoint // '';
  my $actual_endpoint   = $state->{endpoint} // '';

  croak 'Upload state mismatch for endpoint'
    unless $actual_endpoint eq $expected_endpoint;

  croak 'Upload state does not contain an upload_id'
    unless defined $state->{upload_id}
    && length $state->{upload_id};

  $state->{parts} //= {};
}

sub _save_state ($self, $state_file, $state) {
  my $target = path($state_file);
  my $temp   = path("$state_file.tmp.$$");

  $temp->spurt(encode_json($state));

  rename $temp->to_string, $target->to_string
    or croak "Cannot replace upload state $state_file: $!";
}

sub _extract_part ($self, %args) {
  my ($fh, $filename) = tempfile(
    's3-part-XXXXXXXX',
    TMPDIR => 1,
    UNLINK => 0,
  );

  binmode $fh;

  seek $args{source}, $args{offset}, 0
    or croak sprintf(
      'Cannot seek %s to byte %d: %s',
      $args{source_name},
      $args{offset},
      $!,
    );

  my $remaining = $args{length};
  my $buffer;

  while ($remaining > 0) {
    my $wanted = $remaining > 1024 * 1024
      ? 1024 * 1024
      : $remaining;

    my $read = read $args{source}, $buffer, $wanted;

    croak "Cannot read $args{source_name}: $!"
      unless defined $read;

    croak sprintf(
      'Unexpected end of file while reading part %d',
      $args{part_number},
    ) if $read == 0;

    print {$fh} substr($buffer, 0, $read)
      or croak "Cannot write temporary part file: $!";

    $remaining -= $read;
  }

  close $fh
    or croak "Cannot close temporary part file: $!";

  return $filename;
}

sub _remote_part_matches ($self, $part, $expected_size) {
  return 0 unless ref $part eq 'HASH';
  return 0 unless defined $part->{etag};
  return 0 unless defined $part->{size};
  return 0 unless $part->{size} == $expected_size;

  return 1;
}

sub _normalize_part_size ($self, $file_size, $part_size) {
  croak 'part_size must be a positive integer'
    unless defined $part_size
    && $part_size =~ /^\d+$/
    && $part_size > 0;

  my $minimum = 5 * 1024 * 1024;

  # Every part except the final part must be at least 5 MiB.
  $part_size = $minimum
    if $file_size > $part_size && $part_size < $minimum;

  # S3 permits no more than 10,000 parts.
  my $minimum_for_part_limit = ceil($file_size / 10_000);
  $part_size = $minimum_for_part_limit
    if $part_size < $minimum_for_part_limit;

  # Round upward to a whole MiB for predictable part boundaries.
  my $mib = 1024 * 1024;
  $part_size = ceil($part_size / $mib) * $mib;

  my $part_count = ceil($file_size / $part_size);

  croak "Calculated multipart upload has $part_count parts; maximum is 10000"
    if $part_count > 10_000;

  return $part_size;
}

sub _object_url ($self) {
  my $key = $self->key;
  $key =~ s{^/+}{};

  my @segments = split m{/}, $key, -1;

  my $base;

  if (defined $self->endpoint && length $self->endpoint) {
    $base = Mojo::URL->new($self->endpoint);
  }
  elsif ($self->path_style) {
    $base = Mojo::URL->new(
      sprintf 'https://s3.%s.amazonaws.com', $self->region
    );
  }
  else {
    $base = Mojo::URL->new(
      sprintf 'https://%s.s3.%s.amazonaws.com',
        $self->bucket,
        $self->region,
    );
  }

  my @path_parts;

  push @path_parts, $self->bucket
    if $self->path_style;

  push @path_parts, @segments;

  $base->path->parts(\@path_parts);

  return $base;
}

sub _default_state_file ($self, $filename) {
  return "$filename.s3-multipart.json";
}

sub _report_progress ($self, %event) {
  my $callback = $self->on_progress;
  return unless $callback;

  $callback->($self, \%event);
}

sub _validate_configuration ($self) {
  for my $attribute (qw(access_key secret_key region bucket key)) {
    my $value = $self->$attribute;

    croak "$attribute is required"
      unless defined $value && length $value;
  }

  croak 'part_size must be positive'
    unless $self->part_size > 0;

  croak 'max_retries must be zero or greater'
    unless $self->max_retries >= 0;
}

1;

__END__

=head1 NAME

Mojo::S3::ResumableMultipartUpload

=head1 SYNOPSIS

  use Mojo::S3::ResumableMultipartUpload;

  my $uploader = Mojo::S3::ResumableMultipartUpload->new(
    region    => 'us-east-1',
    bucket    => 'my-bucket',
    key       => 'backups/database.tar',
    part_size => 128 * 1024 * 1024,

    on_progress => sub ($uploader, $event) {
      if ($event->{event} eq 'uploaded') {
        printf "Uploaded part %d of %d\n",
          $event->{part_number},
          $event->{part_count};
      }
      elsif ($event->{event} eq 'skipped') {
        printf "Already uploaded part %d of %d\n",
          $event->{part_number},
          $event->{part_count};
      }
    },
  );

  my $result = $uploader->upload('/data/database.tar');

  say "ETag: $result->{etag}";

=head1 DESCRIPTION

Uploads a local file using the Amazon S3 multipart REST API.

The upload ID and completed part information are persisted in a JSON state
file. Rerunning C<upload> with the same source file, bucket, key, part size,
and state file resumes the multipart upload.

Before resuming, the module calls ListParts and treats S3 as authoritative.
Parts whose remote sizes match their expected source-file segments are
skipped.

=head1 CREDENTIALS

The following environment variables are supported:

  AWS_ACCESS_KEY_ID
  AWS_SECRET_ACCESS_KEY
  AWS_SESSION_TOKEN
  AWS_REGION
  AWS_DEFAULT_REGION

The older C<AWS_ACCESS_KEY> and C<AWS_SECRET_KEY> variables accepted by
Mojo::UserAgent::Role::AWSSignature4 are also supported.

=head1 RESUMING

By default, state is written beside the source file:

  database.tar.s3-multipart.json

To select another location:

  $uploader->upload(
    '/data/database.tar',
    state_file => '/var/lib/uploader/database-upload.json',
  );

Do not delete this file until the upload has completed. Once S3 successfully
completes the multipart upload, the state file is removed automatically.

=head1 ABORTING

  my $state = decode_json(
    path('/data/database.tar.s3-multipart.json')->slurp
  );

  $uploader->abort_multipart_upload($state->{upload_id});

Aborting frees the S3 storage associated with incomplete uploaded parts.

=cut

gistfile2.txt

#!/usr/bin/env perl

# for i in {1..3}; do date > $i; done; tar cf test.tar {1..3}
# $ perl ~/repos/tar_simple_index.pl /tmp/test.tar | xargs -d $'\n' -P8 -n1 bash -c 'eval "export $1"; printf "%s\n" "$(date +"[%s.%N] $$... ")" >&2; start=$(date +%s%N); dd if=$TAR_ARCHIVE iflag=skip_bytes,count_bytes skip=$TAR_OFFSET count=$TAR_SIZE status=none | tee /dev/stderr | wc -c; end=$(date +%s%N); s=$(bc -l <<< "scale=9; ($end - $start) / 1000000000"); bps=$(bc <<< "$TAR_SIZE * 8 * 1000000000 / ($end - $start)"); printf "done (%.6f seconds, %'\''d bps)\n" "$s" "$bps" >&2'  bash
# [1784710837.164855260] 840061...
# [1784710837.165632693] 840062...
# [1784710837.165921066] 840063...
# Wed Jul 22 03:16:16 AM CDT 2026
# Wed Jul 22 03:16:24 AM CDT 2026
# Wed Jul 22 03:16:20 AM CDT 2026
# 32
# 32
# 32
# done (0.002138 seconds, 119750 bps)
# done (0.003159 seconds, 81036 bps)
# done (0.002828 seconds, 90508 bps)

use v5.16;
use warnings;
use Fcntl qw(SEEK_CUR SEEK_SET);

my $tar = shift or die "Usage: $0 archive.tar\n";

my $tar_size = -s $tar;
open my $fh, '<:raw', $tar
    or die "$tar: $!";

my $offset = 0;

while (1) {

    my $header;
    my $n = read($fh, $header, 512);
    last unless $n == 512;

    last if $header eq ("\0" x 512);

    my $name   = unpack 'Z100', substr($header,   0,100);
    my $prefix = unpack 'Z155', substr($header, 345,155);

    $name = "$prefix/$name" if length $prefix;

    my $size = oct unpack 'Z12', substr($header,124,12);

    my $data_offset = $offset + 512;
    #last if $data_offset > $tar_size;

    printf "TAR_ARCHIVE=%s TAR_FILENAME=%s TAR_OFFSET=%d TAR_SIZE=%d\n",
        $tar,
        $name,
        $data_offset,
        $size;

    my $next_header = $data_offset + int(($size + 511) / 512) * 512;

    seek $fh, $next_header, SEEK_SET or die "Cannot seek to next header at $next_header: $!";

    $offset = $next_header;
}

gistfile3.txt

#!/usr/bin/env bash
#
# tar-index.sh — print filename, data-start byte, and data length
#
# Usage:
#   tar-index.sh archive.tar
#
# Output:
#   filename<TAB>start_byte<TAB>length
#
# This indexes an uncompressed tar archive. It does not work directly on
# gzip-, bzip2-, xz-, or zstd-compressed tar files because compressed streams
# cannot be randomly addressed using the uncompressed tar offsets.
#

set -euo pipefail

if (( $# != 1 )); then
    printf 'Usage: %s archive.tar\n' "${0##*/}" >&2
    exit 2
fi

tar_file=$1

if [[ ! -r $tar_file ]]; then
    printf 'Error: cannot read %q\n' "$tar_file" >&2
    exit 1
fi

readonly BLOCK_SIZE=512

# Convert a tar numeric field to decimal.
#
# Traditional tar stores sizes as ASCII octal. Some implementations use
# base-256 for very large values; this simple indexer accepts the normal
# octal representation.
parse_octal()
{
    local value=$1

    # Remove spaces and NULs already stripped by the external utilities.
    value=${value//[[:space:]]/}

    if [[ -z $value ]]; then
        printf '0\n'
        return
    fi

    # Force base 8. Prefixing with 10# would incorrectly interpret tar sizes.
    printf '%d\n' "$((8#$value))"
}

# Extract a text field from the current 512-byte header.
#
# Arguments:
#   $1 = header byte offset
#   $2 = field length
read_text_field()
{
    local offset=$1
    local length=$2

    dd if="$tar_file" \
        iflag=skip_bytes,count_bytes \
        skip=$((header_offset + offset)) \
        count="$length" \
        status=none |
        tr -d '\000'
}

# Return success when the current 512-byte header consists entirely of zeros.
is_zero_header()
{
    local nonzero_count

    nonzero_count=$(
        dd if="$tar_file" \
            iflag=skip_bytes,count_bytes \
            skip="$header_offset" \
            count="$BLOCK_SIZE" \
            status=none |
            tr -d '\000' |
            wc -c
    )

    (( nonzero_count == 0 ))
}

archive_size=$(stat --format='%s' -- "$tar_file")
header_offset=0

while (( header_offset + BLOCK_SIZE <= archive_size )); do
    if is_zero_header; then
        break
    fi

    name=$(read_text_field 0 100)
    size_field=$(read_text_field 124 12)
    typeflag=$(read_text_field 156 1)
    prefix=$(read_text_field 345 155)

    size=$(parse_octal "$size_field")
    data_start=$((header_offset + BLOCK_SIZE))
    data_end=$((data_start + size))

    if [[ -n $prefix ]]; then
        name=$prefix/$name
    fi

    # Print ordinary files and entries whose type field is empty.
    #
    # Type "0" is a normal file.
    # An empty type field is also historically treated as a normal file.
    if [[ -z $typeflag || $typeflag == 0 ]]; then
        printf '%s\t%d\t%d\n' "$name" "$data_start" "$size"
    fi

    # Tar members are padded to the next 512-byte boundary.
    padded_size=$(( (size + BLOCK_SIZE - 1) / BLOCK_SIZE * BLOCK_SIZE ))
    header_offset=$((data_start + padded_size))
done

gistfile4.txt

#!/usr/bin/env perl

use v5.42;
use warnings;

use Fcntl qw(SEEK_SET);

my $PROGRAM_VERSION = '1.0';
my $BLOCK_SIZE      = 512;

my $archive = shift
    or die "Usage: $0 archive.tar\n";

open my $tar, '<:raw', $archive
    or die "Cannot open '$archive': $!\n";

sub trim_tar_text ($value)
{
    $value =~ s/\0.*\z//s;
    $value =~ s/\s+\z//;
    return $value;
}

sub parse_tar_number ($value)
{
    # GNU/base-256 numeric encoding.
    if (length($value) && (ord(substr($value, 0, 1)) & 0x80)) {
        my @bytes = unpack 'C*', $value;

        # Clear the base-256 marker.
        $bytes[0] &= 0x7f;

        my $number = 0;
        $number = ($number << 8) | $_ for @bytes;

        return $number;
    }

    # Traditional ASCII-octal encoding.
    $value =~ s/\0.*\z//s;
    $value =~ s/^\s+|\s+$//g;

    return 0 unless length $value;

    die "Invalid tar numeric field: '$value'\n"
        unless $value =~ /\A[0-7]+\z/;

    return oct $value;
}

sub shell_quote ($value)
{
    $value //= '';

    # POSIX-shell-safe single quoting:
    #
    #   abc'def  ->  'abc'"'"'def'
    $value =~ s/'/'"'"'/g;

    return "'$value'";
}

sub tar_format ($header)
{
    my $magic   = substr($header, 257, 6);
    my $version = substr($header, 263, 2);

    return 'v7'
        unless $magic =~ /\Austar/;

    return 'gnu'
        if $magic eq "ustar " && $version eq " \0";

    return 'ustar'
        if $magic eq "ustar\0" && $version eq '00';

    return 'ustar';
}

sub tar_filetype ($typeflag)
{
    return 'f' if $typeflag eq "\0" || $typeflag eq '0';
    return 'h' if $typeflag eq '1';
    return 'l' if $typeflag eq '2';
    return 'c' if $typeflag eq '3';
    return 'b' if $typeflag eq '4';
    return 'd' if $typeflag eq '5';

    # GNU tar --to-command currently invokes the command only for regular
    # files, but retain a useful value for unsupported/special types.
    return $typeflag;
}

sub emit_record (%env)
{
    my @order = qw(
        TAR_VERSION
        TAR_ARCHIVE
        TAR_BLOCKING_FACTOR
        TAR_VOLUME
        TAR_FORMAT
        TAR_SUBCOMMAND
        TAR_FILETYPE
        TAR_MODE
        TAR_FILENAME
        TAR_REALNAME
        TAR_UNAME
        TAR_GNAME
        TAR_ATIME
        TAR_MTIME
        TAR_CTIME
        TAR_SIZE
        TAR_UID
        TAR_GID
        TAR_OFFSET
    );

    my $record = join ' ',
        map { "$_=" . shell_quote($env{$_}) } @order;

    print STDOUT $record, "\0";
}

my $header_offset = 0;
my $zero_headers  = 0;

while (1) {
    my $header = '';

    my $bytes_read = read $tar, $header, $BLOCK_SIZE;

    die "Error reading '$archive': $!\n"
        unless defined $bytes_read;

    last if $bytes_read == 0;

    die "Truncated tar header at byte $header_offset\n"
        unless $bytes_read == $BLOCK_SIZE;

    if ($header eq "\0" x $BLOCK_SIZE) {
        last if ++$zero_headers == 2;

        $header_offset += $BLOCK_SIZE;
        next;
    }

    $zero_headers = 0;

    my $name = trim_tar_text(substr($header, 0,   100));
    my $mode = trim_tar_text(substr($header, 100, 8));
    my $uid  = parse_tar_number(substr($header, 108, 8));
    my $gid  = parse_tar_number(substr($header, 116, 8));
    my $size = parse_tar_number(substr($header, 124, 12));
    my $mtime = parse_tar_number(substr($header, 136, 12));

    my $typeflag = substr($header, 156, 1);
    my $uname    = trim_tar_text(substr($header, 265, 32));
    my $gname    = trim_tar_text(substr($header, 297, 32));
    my $prefix   = trim_tar_text(substr($header, 345, 155));

    $name = "$prefix/$name"
        if length $prefix;

    my $data_offset = $header_offset + $BLOCK_SIZE;
    my $filetype    = tar_filetype($typeflag);

    # GNU tar --to-command receives regular-file contents on standard input.
    # Therefore, emit work records only for regular files.
    if ($filetype eq 'f') {
        emit_record(
            TAR_VERSION         => "tar-index.pl $PROGRAM_VERSION",
            TAR_ARCHIVE         => $archive,
            TAR_BLOCKING_FACTOR => 20,
            TAR_VOLUME          => 1,
            TAR_FORMAT          => tar_format($header),
            TAR_SUBCOMMAND      => '-x',

            TAR_FILETYPE => $filetype,
            TAR_MODE     => $mode,
            TAR_FILENAME => $name,
            TAR_REALNAME => $name,
            TAR_UNAME    => $uname,
            TAR_GNAME    => $gname,

            # Basic ustar headers contain mtime but not atime or ctime.
            TAR_ATIME => '',
            TAR_MTIME => $mtime,
            TAR_CTIME => '',

            TAR_SIZE => $size,
            TAR_UID  => $uid,
            TAR_GID  => $gid,

            # Custom variable: absolute byte offset of the member data.
            TAR_OFFSET => $data_offset,
        );
    }

    my $padded_size =
        int(($size + $BLOCK_SIZE - 1) / $BLOCK_SIZE) * $BLOCK_SIZE;

    my $next_header = $data_offset + $padded_size;

    sysseek $tar, $next_header, SEEK_SET
        or die "Cannot seek to byte $next_header in '$archive': $!\n";

    $header_offset = $next_header;
}

gistfile5.txt

$ function expand_tar { local tarball="$1" callback="${2:-wc}"; test -n "$tarball" || { echo "Usage: $FUNCNAME tarball"; return 1; }; perl -MFcntl=SEEK_SET -E 'my $tar = shift or die "Usage: $0 archive.tar\n";my $callback=shift; my $tar_size = -s $tar;open my $fh, q(<:raw), $tar or die "$tar: $!"; my $offset = 0;while (1) {    my $header;    my $n = read($fh, $header, 512); last unless $n == 512;    last if $header eq ("\0" x 512);    my $name   = unpack q(Z100), substr($header,   0,100); my $prefix = unpack q(Z155), substr($header, 345,155); $name = "$prefix/$name" if length $prefix;    my $size = oct unpack q(Z12), substr($header,124,12); my $data_offset = $offset + 512; printf "TAR_ARCHIVE=%s TAR_CALLBACK=\"%s\" TAR_FILENAME=%s TAR_OFFSET=%d TAR_SIZE=%d\n",        $tar,        $callback, $name,        $data_offset,        $size;    my $next_header = $data_offset + int(($size + 511) / 512) * 512;    seek $fh, $next_header, SEEK_SET or die "Cannot seek to next header at $next_header: $!";    $offset = $next_header;}' "$tarball" "$callback" | xargs -d $'\n' -P2 -n1 bash -c 'eval "export $1"; printf "%s\n" "$(date +"[%s.%N] $$... ")" >&2; start=$(date +%s%N); dd if=$TAR_ARCHIVE iflag=skip_bytes,count_bytes skip=$TAR_OFFSET count=$TAR_SIZE status=none | ${TAR_CALLBACK[@]}; end=$(date +%s%N); s=$(bc -l <<< "scale=9; ($end - $start) / 1000000000"); bps=$(bc <<< "$TAR_SIZE * 8 * 1000000000 / ($end - $start)"); printf "done (%.6f seconds, %'\''d bps)\n" "$s" "$bps" >&2'  bash ;}
$ expand_tar /tmp/test.tar "wc -c"
[1784711860.676171330] 840880...
[1784711860.676781431] 840881...
32
32
done (0.004564 seconds, 56,086 bps)
done (0.004275 seconds, 59,889 bps)
[1784711860.693518758] 840900...
32
done (0.002409 seconds, 106,279 bps)

upload.pl

#!/usr/bin/env perl

use Mojo::Base -strict, -signatures;
use lib 'lib';

use Mojo::S3::ResumableMultipartUpload;

my ($filename, $bucket, $key) = @ARGV;

die "Usage: $0 FILE BUCKET KEY\n"
  unless defined $key;

my $uploader = Mojo::S3::ResumableMultipartUpload->new(
  bucket    => $bucket,
  key       => $key,
  region    => $ENV{AWS_REGION} // 'us-east-1',
  part_size => 128 * 1024 * 1024,

  on_progress => sub ($uploader, $event) {
    return unless $event->{event} =~ /^(?:uploaded|skipped)$/;

    my $percent = 100 * (
      ($event->{offset} + $event->{bytes}) / $event->{file_size}
    );

    printf STDERR "%-8s part %d/%d — %.1f%%\n",
      uc($event->{event}),
      $event->{part_number},
      $event->{part_count},
      $percent;
  },
);

my $result = $uploader->upload($filename);

printf "Uploaded s3://%s/%s\nETag: %s\n",
  $result->{bucket},
  $result->{key},
  $result->{etag} // '(not returned)';