lib/File/Rsync/Mirror/Recentfile.pm

   1 package File::Rsync::Mirror::Recentfile;
   2
   3 # use warnings;
   4 use strict;
   5
   6 =encoding utf-8
   7
   8 =head1 NAME
   9
  10 File::Rsync::Mirror::Recentfile - mirroring via rsync made efficient
  11
  12 =cut
  13
  14 my $HAVE = {};
  15 for my $package (
  16                  "Data::Serializer",
  17                  "File::Rsync"
  18                 ) {
  19     $HAVE->{$package} = eval qq{ require $package; };
  20 }
  21 use Config;
  22 use File::Basename qw(basename dirname fileparse);
  23 use File::Copy qw(cp);
  24 use File::Path qw(mkpath);
  25 use File::Rsync::Mirror::Recentfile::FakeBigFloat qw(:all);
  26 use File::Temp;
  27 use List::Util qw(first max min);
  28 use Scalar::Util qw(reftype);
  29 use Storable;
  30 use Time::HiRes qw();
  31 use YAML::Syck;
  32
  33 use version; our $VERSION = qv('0.0.1');
  34
  35 use constant MAX_INT => ~0>>1; # anything better?
  36 use constant DEFAULT_PROTOCOL => 1;
  37
  38 # cf. interval_secs
  39 my %seconds;
  40
  41 # maybe subclass if this mapping is bad?
  42 my %serializers;
  43
  44 =head1 SYNOPSIS
  45
  46 B<!!!! PRE-ALPHA ALERT !!!!>
  47
  48 Nothing in here is believed to be stable, nothing yet intended for
  49 public consumption. The plan is to provide scripts that act as
  50 frontends for all the backend functionality. Option and method names
  51 will very likely change.
  52
  53 For the rationale see the section BACKGROUND.
  54
  55 This is published only for developers of the (yet to be named)
  56 script(s).
  57
  58 Writer (of a single file):
  59
  60     use File::Rsync::Mirror::Recentfile;
  61     my $fr = File::Rsync::Mirror::Recentfile->new
  62       (
  63        interval => q(6h),
  64        filenameroot => "RECENT",
  65        comment => "These 'RECENT' files are part of a test of a new CPAN mirroring concept. Please ignore them for now.",
  66        localroot => "/home/ftp/pub/PAUSE/authors/",
  67        aggregator => [qw(1d 1W 1M 1Q 1Y Z)],
  68       );
  69     $rf->update("/home/ftp/pub/PAUSE/authors/id/A/AN/ANDK/CPAN-1.92_63.tar.gz","new");
  70
  71 Reader/mirrorer:
  72
  73     my $rf = File::Rsync::Mirror::Recentfile->new
  74       (
  75        filenameroot => "RECENT",
  76        ignore_link_stat_errors => 1,
  77        interval => q(6h),
  78        localroot => "/home/ftp/pub/PAUSE/authors",
  79        remote_dir => "",
  80        remote_host => "pause.perl.org",
  81        remote_module => "authors",
  82        rsync_options => {
  83                          compress => 1,
  84                          'rsync-path' => '/usr/bin/rsync',
  85                          links => 1,
  86                          times => 1,
  87                          'omit-dir-times' => 1,
  88                          checksum => 1,
  89                         },
  90        verbose => 1,
  91       );
  92     $rf->mirror;
  93
  94 Aggregator (usually the writer):
  95
  96     my $rf = File::Rsync::Mirror::Recentfile->new_from_file ( $file );
  97     $rf->aggregate;
  98
  99 =head1 EXPORT
 100
 101 No exports.
 102
 103 =head1 CONSTRUCTORS / DESTRUCTOR
 104
 105 =head2 my $obj = CLASS->new(%hash)
 106
 107 Constructor. On every argument pair the key is a method name and the
 108 value is an argument to that method name.
 109
 110 If a recentfile for this resource already exists, metadata that are
 111 not defined by the constructor will be fetched from there as soon as
 112 it is being read by recent_events().
 113
 114 =cut
 115
 116 sub new {
 117     my($class, @args) = @_;
 118     my $self = bless {}, $class;
 119     while (@args) {
 120         my($method,$arg) = splice @args, 0, 2;
 121         $self->$method($arg);
 122     }
 123     unless (defined $self->protocol) {
 124         $self->protocol(DEFAULT_PROTOCOL);
 125     }
 126     unless (defined $self->filenameroot) {
 127         $self->filenameroot("RECENT");
 128     }
 129     unless (defined $self->serializer_suffix) {
 130         $self->serializer_suffix(".yaml");
 131     }
 132     return $self;
 133 }
 134
 135 =head2 my $obj = CLASS->new_from_file($file)
 136
 137 Constructor. $file is a I<recentfile>.
 138
 139 =cut
 140
 141 sub new_from_file {
 142     my($class, $file) = @_;
 143     my $self = bless {}, $class;
 144     $self->_rfile($file);
 145     #?# $self->lock;
 146     my $serialized = do { open my $fh, $file or die "Could not open '$file': $!";
 147                            local $/;
 148                            <$fh>;
 149                        };
 150     # XXX: we can skip this step when the metadata are sufficient, but
 151     # we cannot parse the file without some magic stuff about
 152     # serialized formats
 153     while (-l $file) {
 154         my($name,$path) = fileparse $file;
 155         my $symlink = readlink $file;
 156         if ($symlink =~ m|/|) {
 157             die "FIXME: filenames containing '/' not supported, got $symlink";
 158         }
 159         $file = File::Spec->catfile ( $path, $symlink );
 160     }
 161     my($name,$path,$suffix) = fileparse $file, keys %serializers;
 162     $self->serializer_suffix($suffix);
 163     $self->localroot($path);
 164     die "Could not determine file format from suffix" unless $suffix;
 165     my $deserialized;
 166     if ($suffix eq ".yaml") {
 167         require YAML::Syck;
 168         $deserialized = YAML::Syck::LoadFile($file);
 169     } elsif ($HAVE->{"Data::Serializer"}) {
 170         my $serializer = Data::Serializer->new
 171             ( serializer => $serializers{$suffix} );
 172         $deserialized = $serializer->raw_deserialize($serialized);
 173     } else {
 174         die "Data::Serializer not installed, cannot proceed with suffix '$suffix'";
 175     }
 176     while (my($k,$v) = each %{$deserialized->{meta}}) {
 177         next if $k ne lc $k; # "Producers"
 178         $self->$k($v);
 179     }
 180     unless (defined $self->protocol) {
 181         $self->protocol(DEFAULT_PROTOCOL);
 182     }
 183     return $self;
 184 }
 185
 186 =head2 DESTROY
 187
 188 A simple unlock.
 189
 190 =cut
 191 sub DESTROY { shift->unlock }
 192
 193 =head1 ACCESSORS
 194
 195 =cut
 196
 197 my @accessors;
 198
 199 BEGIN {
 200     @accessors = (
 201                   "_current_tempfile",
 202                   "_current_tempfile_fh",
 203                   "_delayed_operations",
 204                   "_done",
 205                   "_interval",
 206                   "_is_locked",
 207                   "_localroot",
 208                   "_merged",
 209                   "_pathdb",
 210                   "_remember_last_uptodate_call",
 211                   "_remote_dir",
 212                   "_remoteroot",
 213                   "_rfile",
 214                   "_rsync",
 215                   "_seeded",
 216                   "_uptodateness_ever_reached",
 217                   "_use_tempfile",
 218                  );
 219
 220     my @pod_lines =
 221         split /\n/, <<'=cut'; push @accessors, grep {s/^=item\s+//} @pod_lines; }
 222
 223 =over 4
 224
 225 =item aggregator
 226
 227 A list of interval specs that tell the aggregator which I<recentfile>s
 228 are to be produced.
 229
 230 =item canonize
 231
 232 The name of a method to canonize the path before rsyncing. Only
 233 supported value is C<naive_path_normalize>. Defaults to that.
 234
 235 =item comment
 236
 237 A comment about this tree and setup.
 238
 239 =item dirtymark
 240
 241 A timestamp. The dirtymark is updated whenever an out of band change
 242 on the origin server is performed that violates the protocol. Say,
 243 they add or remove files in the middle somewhere. Slaves must react
 244 with a devaluation of their C<done> structure which then leads to a
 245 full re-sync of all files.
 246
 247 =item filenameroot
 248
 249 The (prefix of the) filename we use for this I<recentfile>. Defaults to
 250 C<RECENT>. The string must not contain a directory separator.
 251
 252 =item have_mirrored
 253
 254 Timestamp remembering when we mirrored this recentfile the last time.
 255 Only relevant for slaves.
 256
 257 =item ignore_link_stat_errors
 258
 259 If set to true, rsync errors are ignored that complain about link stat
 260 errors. These seem to happen only when there are files missing at the
 261 origin. In race conditions this can always happen, so it is
 262 recommended to set this value to true.
 263
 264 =item is_slave
 265
 266 If set to true, this object will fetch a new recentfile from remote
 267 when the timespan between the last mirror (see have_mirrored) and now
 268 is too large (currently hardcoded arbitrary 420 seconds).
 269
 270 =item locktimeout
 271
 272 After how many seconds shall we die if we cannot lock a I<recentfile>?
 273 Defaults to 600 seconds.
 274
 275 =item loopinterval
 276
 277 When mirror_loop is called, this accessor can specify how much time
 278 every loop shall at least take. If the work of a loop is done before
 279 that time has gone, sleeps for the rest of the time. Defaults to
 280 arbitrary 42 seconds.
 281
 282 =item max_files_per_connection
 283
 284 Maximum number of files that are transferred on a single rsync call.
 285 Setting it higher means higher performance at the price of holding
 286 connections longer and potentially disturbing other users in the pool.
 287 Defaults to the arbitrary value 42.
 288
 289 =item max_rsync_errors
 290
 291 When rsync operations encounter that many errors without any resetting
 292 success in between, then we die. Defaults to unlimited. A value of
 293 -1 means we run forever ignoring all rsync errors.
 294
 295 =item minmax
 296
 297 Hashref remembering when we read the recent_events from this file the
 298 last time and what the timespan was.
 299
 300 =item protocol
 301
 302 When the RECENT file format changes, we increment the protocol. We try
 303 to support older protocols in later releases.
 304
 305 =item remote_host
 306
 307 The host we are mirroring from. Leave empty for the local filesystem.
 308
 309 =item remote_module
 310
 311 Rsync servers have so called modules to separate directory trees from
 312 each other. Put here the name of the module under which we are
 313 mirroring. Leave empty for local filesystem.
 314
 315 =item rsync_options
 316
 317 Things like compress, links, times or checksums. Passed in to the
 318 File::Rsync object used to run the mirror.
 319
 320 =item serializer_suffix
 321
 322 Mostly untested accessor. The only well tested format for
 323 I<recentfile>s at the moment is YAML. It is used with YAML::Syck via
 324 Data::Serializer. But in principle other formats are supported as
 325 well. See section SERIALIZERS below.
 326
 327 =item sleep_per_connection
 328
 329 Sleep that many seconds (floating point OK) after every chunk of rsyncing
 330 has finished. Defaults to arbitrary 0.42.
 331
 332 =item ttl
 333
 334 Time to live. Number of seconds after which this recentfile must be
 335 fetched again from the origin server. Only relevant for slaves.
 336 Defaults to arbitrary 24.2 seconds.
 337
 338 =item verbose
 339
 340 Boolean to turn on a bit verbosity.
 341
 342 =back
 343
 344 =cut
 345
 346 use accessors @accessors;
 347
 348 =head1 METHODS
 349
 350 =head2 (void) $obj->aggregate( %options )
 351
 352 Takes all intervals that are collected in the accessor called
 353 aggregator. Sorts them by actual length of the interval.
 354 Removes those that are shorter than our own interval. Then merges this
 355 object into the next larger object. The merging continues upwards
 356 as long as the next I<recentfile> is old enough to warrant a merge.
 357
 358 If a merge is warranted is decided according to the interval of the
 359 previous interval so that larger files are not so often updated as
 360 smaller ones. If $options{force} is true, all files get updated.
 361
 362 Here is an example to illustrate the behaviour. Given aggregators
 363
 364   1h 1d 1W 1M 1Q 1Y Z
 365
 366 then
 367
 368   1h updates 1d on every call to aggregate()
 369   1d updates 1W earliest after 1h
 370   1W updates 1M earliest after 1d
 371   1M updates 1Q earliest after 1W
 372   1Q updates 1Y earliest after 1M
 373   1Y updates  Z earliest after 1Q
 374
 375 Note that all but the smallest recentfile get updated at an arbitrary
 376 rate and as such are quite useless on their own.
 377
 378 =cut
 379
 380 sub aggregate {
 381     my($self, %option) = @_;
 382     my @aggs = sort { $a->{secs} <=> $b->{secs} }
 383         grep { $_->{secs} >= $self->interval_secs }
 384             map { { interval => $_, secs => $self->interval_secs($_)} }
 385                 $self->interval, @{$self->aggregator || []};
 386     $self->update;
 387     $aggs[0]{object} = $self;
 388   AGGREGATOR: for my $i (0..$#aggs-1) {
 389         my $this = $aggs[$i]{object};
 390         my $next = $this->_sparse_clone;
 391         $next->interval($aggs[$i+1]{interval});
 392         my $want_merge = 0;
 393         if ($option{force} || $i == 0) {
 394             $want_merge = 1;
 395         } else {
 396             my $next_rfile = $next->rfile;
 397             if (-e $next_rfile) {
 398                 my $prev = $aggs[$i-1]{object};
 399                 local $^T = time;
 400                 my $next_age = 86400 * -M $next_rfile;
 401                 if ($next_age > $prev->interval_secs) {
 402                     $want_merge = 1;
 403                 }
 404             } else {
 405                 $want_merge = 1;
 406             }
 407         }
 408         if ($want_merge) {
 409             $next->merge($this);
 410             $aggs[$i+1]{object} = $next;
 411         } else {
 412             last AGGREGATOR;
 413         }
 414     }
 415 }
 416
 417 # collect file size and mtime for all files of this aggregate
 418 sub _debug_aggregate {
 419     my($self) = @_;
 420     my @aggs = sort { $a->{secs} <=> $b->{secs} }
 421         map { { interval => $_, secs => $self->interval_secs($_)} }
 422             $self->interval, @{$self->aggregator || []};
 423     my $report = [];
 424     for my $i (0..$#aggs) {
 425         my $this = Storable::dclone $self;
 426         $this->interval($aggs[$i]{interval});
 427         my $rfile = $this->rfile;
 428         my @stat = stat $rfile;
 429         push @$report, {rfile => $rfile, size => $stat[7], mtime => $stat[9]};
 430     }
 431     $report;
 432 }
 433
 434 # (void) $self->_assert_symlink()
 435 sub _assert_symlink {
 436     my($self) = @_;
 437     my $recentrecentfile = File::Spec->catfile
 438         (
 439          $self->localroot,
 440          sprintf
 441          (
 442           "%s.recent",
 443           $self->filenameroot
 444          )
 445         );
 446     if ($Config{d_symlink} eq "define") {
 447         my $howto_create_symlink; # 0=no need; 1=straight symlink; 2=rename symlink
 448         if (-l $recentrecentfile) {
 449             my $found_symlink = readlink $recentrecentfile;
 450             if ($found_symlink eq $self->rfilename) {
 451                 return;
 452             } else {
 453                 $howto_create_symlink = 2;
 454             }
 455         } else {
 456             $howto_create_symlink = 1;
 457         }
 458         if (1 == $howto_create_symlink) {
 459             symlink $self->rfilename, $recentrecentfile or die "Could not create symlink '$recentrecentfile': $!"
 460         } else {
 461             unlink "$recentrecentfile.$$"; # may fail
 462             symlink $self->rfilename, "$recentrecentfile.$$" or die "Could not create symlink '$recentrecentfile.$$': $!";
 463             rename "$recentrecentfile.$$", $recentrecentfile or die "Could not rename '$recentrecentfile.$$' to $recentrecentfile: $!";
 464         }
 465     } else {
 466         warn "Warning: symlinks not supported on this system, doing a copy instead\n";
 467         unlink "$recentrecentfile.$$"; # may fail
 468         cp $self->rfilename, "$recentrecentfile.$$" or die "Could not copy to '$recentrecentfile.$$': $!";
 469         rename "$recentrecentfile.$$", $recentrecentfile or die "Could not rename '$recentrecentfile.$$' to $recentrecentfile: $!";
 470     }
 471 }
 472
 473 =head2 $hashref = $obj->delayed_operations
 474
 475 A hash of hashes containing unlink and rmdir operations which had to
 476 wait until the recentfile got unhidden in order to not confuse
 477 downstream mirrors (in case we have some).
 478
 479 =cut
 480
 481 sub delayed_operations {
 482     my($self) = @_;
 483     my $x = $self->_delayed_operations;
 484     unless (defined $x) {
 485         $x = {
 486               unlink => {},
 487               rmdir => {},
 488              };
 489         $self->_delayed_operations ($x);
 490     }
 491     return $x;
 492 }
 493
 494 =head2 $done = $obj->done
 495
 496 $done is a reference to a File::Rsync::Mirror::Recentfile::Done object
 497 that keeps track of rsync activities. Only needed and used when we are
 498 a mirroring slave.
 499
 500 =cut
 501
 502 sub done {
 503     my($self) = @_;
 504     my $done = $self->_done;
 505     if (!$done) {
 506         require File::Rsync::Mirror::Recentfile::Done;
 507         $done = File::Rsync::Mirror::Recentfile::Done->new();
 508         $done->_rfinterval ($self->interval);
 509         $self->_done ( $done );
 510     }
 511     return $done;
 512 }
 513
 514 =head2 $tempfilename = $obj->get_remote_recentfile_as_tempfile ()
 515
 516 Stores the remote I<recentfile> locally as a tempfile. The caller is
 517 responsible to remove the file after use.
 518
 519 Note: if you're intending to act as an rsync server for other slaves,
 520 then you must prefer this method to fetch that file with
 521 get_remotefile(). Otherwise downstream mirrors would expect you to
 522 already have mirrored all the files that are in the I<recentfile>
 523 before you have them mirrored.
 524
 525 =cut
 526
 527 sub get_remote_recentfile_as_tempfile {
 528     my($self) = @_;
 529     mkpath $self->localroot;
 530     my $fh;
 531     my $trfilename;
 532     if ( $self->_use_tempfile() ) {
 533         return $self->_current_tempfile if ! $self->ttl_reached;
 534         $fh = $self->_current_tempfile_fh;
 535         $trfilename = $self->rfilename;
 536     } else {
 537         $trfilename = $self->rfilename;
 538     }
 539
 540     my $dst;
 541     if ($fh) {
 542         $dst = $self->_current_tempfile;
 543     } else {
 544         $fh = $self->_get_remote_rat_provide_tempfile_object ($trfilename);
 545         $dst = $fh->filename;
 546         $self->_current_tempfile ($dst);
 547         my $rfile = eval { $self->rfile; }; # may fail (RECENT.recent has no rfile)
 548         if (defined $rfile && -e $rfile) {
 549             # saving on bandwidth. Might need to be configurable
 550             # $self->bandwidth_is_cheap?
 551             cp $rfile, $dst or die "Could not copy '$rfile' to '$dst': $!"
 552         }
 553     }
 554     my $src = join ("/",
 555                     $self->remoteroot,
 556                     $trfilename,
 557                    );
 558     if ($self->verbose) {
 559         my $doing = -e $dst ? "Sync" : "Get";
 560         my $display_dst = join "/", "...", basename(dirname($dst)), basename($dst);
 561         printf STDERR
 562             (
 563              "%-4s %d (1/1/%s) temp %s ... ",
 564              $doing,
 565              time,
 566              $self->interval,
 567              $display_dst,
 568             );
 569     }
 570     my $gaveup = 0;
 571     my $retried = 0;
 572     while (!$self->rsync->exec(
 573                                src => $src,
 574                                dst => $dst,
 575                               )) {
 576         $self->register_rsync_error ($self->rsync->err);
 577         if (++$retried >= 3) {
 578             warn "XXX giving up";
 579             $gaveup = 1;
 580             last;
 581         }
 582     }
 583     if ($gaveup) {
 584         printf STDERR "Warning: gave up mirroring %s, will try again later", $self->interval;
 585     } else {
 586         $self->_refresh_internals ($dst);
 587         $self->have_mirrored (Time::HiRes::time);
 588         $self->un_register_rsync_error ();
 589     }
 590     if ($self->verbose) {
 591         print STDERR "DONE\n";
 592     }
 593     my $mode = 0644;
 594     chmod $mode, $dst or die "Could not chmod $mode '$dst': $!";
 595     return $dst;
 596 }
 597
 598 sub _get_remote_rat_provide_tempfile_object {
 599     my($self, $trfilename) = @_;
 600     my $fh = File::Temp->new
 601         (TEMPLATE => sprintf(".FRMRecent-%s-XXXX",
 602                              $trfilename,
 603                             ),
 604          DIR => $self->localroot,
 605          SUFFIX => $self->serializer_suffix,
 606          UNLINK => $self->_use_tempfile,
 607         );
 608     if ($self->_use_tempfile) {
 609         $self->_current_tempfile_fh ($fh); # delay self destruction
 610     }
 611     return $fh;
 612 }
 613
 614 =head2 $localpath = $obj->get_remotefile ( $relative_path )
 615
 616 Rsyncs one single remote file to local filesystem.
 617
 618 Note: no locking is done on this file. Any number of processes may
 619 mirror this object.
 620
 621 Note II: do not use for recentfiles. If you are a cascading
 622 slave/server combination, it would confuse other slaves. They would
 623 expect the contents of these recentfiles to be available. Use
 624 get_remote_recentfile_as_tempfile() instead.
 625
 626 =cut
 627
 628 sub get_remotefile {
 629     my($self, $path) = @_;
 630     my $dst = File::Spec->catfile($self->localroot, $path);
 631     mkpath dirname $dst;
 632     if ($self->verbose) {
 633         my $doing = -e $dst ? "Sync" : "Get";
 634         printf STDERR
 635             (
 636              "%-4s %d (1/1/%s) %s ... ",
 637              $doing,
 638              time,
 639              $self->interval,
 640              $path,
 641             );
 642     }
 643     while (!$self->rsync->exec(
 644                                src => join("/",
 645                                            $self->remoteroot,
 646                                            $path),
 647                                dst => $dst,
 648                               )) {
 649         $self->register_rsync_error ($self->rsync->err);
 650     }
 651     $self->un_register_rsync_error ();
 652     if ($self->verbose) {
 653         print STDERR "DONE\n";
 654     }
 655     return $dst;
 656 }
 657
 658 =head2 $obj->interval ( $interval_spec )
 659
 660 Get/set accessor. $interval_spec is a string and described below in
 661 the section INTERVAL SPEC.
 662
 663 =cut
 664
 665 sub interval {
 666     my ($self, $interval) = @_;
 667     if (@_ >= 2) {
 668         $self->_interval($interval);
 669         $self->_rfile(undef);
 670     }
 671     $interval = $self->_interval;
 672     unless (defined $interval) {
 673         # do not ask the $self too much, it recurses!
 674         require Carp;
 675         Carp::confess("Alert: interval undefined for '".$self."'. Cannot continue.");
 676     }
 677     return $interval;
 678 }
 679
 680 =head2 $secs = $obj->interval_secs ( $interval_spec )
 681
 682 $interval_spec is described below in the section INTERVAL SPEC. If
 683 empty defaults to the inherent interval for this object.
 684
 685 =cut
 686
 687 sub interval_secs {
 688     my ($self, $interval) = @_;
 689     $interval ||= $self->interval;
 690     unless (defined $interval) {
 691         die "interval_secs() called without argument on an object without a declared one";
 692     }
 693     my ($n,$t) = $interval =~ /^(\d*)([smhdWMQYZ]$)/ or
 694         die "Could not determine seconds from interval[$interval]";
 695     if ($interval eq "Z") {
 696         return MAX_INT;
 697     } elsif (exists $seconds{$t} and $n =~ /^\d+$/) {
 698         return $seconds{$t}*$n;
 699     } else {
 700         die "Invalid interval specification: n[$n]t[$t]";
 701     }
 702 }
 703
 704 =head2 $obj->localroot ( $localroot )
 705
 706 Get/set accessor. The local root of the tree.
 707
 708 =cut
 709
 710 sub localroot {
 711     my ($self, $localroot) = @_;
 712     if (@_ >= 2) {
 713         $self->_localroot($localroot);
 714         $self->_rfile(undef);
 715     }
 716     $localroot = $self->_localroot;
 717 }
 718
 719 =head2 $ret = $obj->local_path($path_found_in_recentfile)
 720
 721 Combines the path to our local mirror and the path of an object found
 722 in this I<recentfile>. In other words: the target of a mirror operation.
 723
 724 Implementation note: We split on slashes and then use
 725 File::Spec::catfile to adjust to the local operating system.
 726
 727 =cut
 728
 729 sub local_path {
 730     my($self,$path) = @_;
 731     unless (defined $path) {
 732         # seems like a degenerated case
 733         return $self->localroot;
 734     }
 735     my @p = split m|/|, $path;
 736     File::Spec->catfile($self->localroot,@p);
 737 }
 738
 739 =head2 (void) $obj->lock
 740
 741 Locking is implemented with an C<mkdir> on a locking directory
 742 (C<.lock> appended to $rfile).
 743
 744 =cut
 745
 746 sub lock {
 747     my ($self) = @_;
 748     # not using flock because it locks on filehandles instead of
 749     # old school ressources.
 750     my $locked = $self->_is_locked and return;
 751     my $rfile = $self->rfile;
 752     # XXX need a way to allow breaking the lock
 753     my $start = time;
 754     my $locktimeout = $self->locktimeout || 600;
 755     while (not mkdir "$rfile.lock") {
 756         Time::HiRes::sleep 0.01;
 757         if (time - $start > $locktimeout) {
 758             die "Could not acquire lockdirectory '$rfile.lock': $!";
 759         }
 760     }
 761     $self->_is_locked (1);
 762 }
 763
 764 =head2 (void) $obj->merge ($other)
 765
 766 Bulk update of this object with another one. It's used to merge a
 767 smaller and younger $other object into the current one. If this file
 768 is a C<Z> file, then we do not merge in objects of type C<delete>. But
 769 if we encounter an object of type delete we delete the corresponding
 770 C<new> object if we have it.
 771
 772 If there is nothing to be merged, nothing is done.
 773
 774 =cut
 775
 776 sub merge {
 777     my($self, $other) = @_;
 778     $self->_merge_sanitycheck ( $other );
 779     $other->lock;
 780     my $other_recent = $other->recent_events || [];
 781     # $DB::single++ if $other->interval_secs eq "2" and grep {$_->{epoch} eq "999.999"} @$other_recent;
 782     $self->lock;
 783     my $my_recent = $self->recent_events || [];
 784
 785     # calculate the target time span
 786     my $myepoch = $my_recent->[0] ? $my_recent->[0]{epoch} : undef;
 787     my $epoch = $other_recent->[0] ? $other_recent->[0]{epoch} : $myepoch;
 788     my $oldest_allowed = 0;
 789     my $something_done;
 790     unless ($my_recent->[0]) {
 791         # obstetrics
 792         $something_done = 1;
 793     }
 794     if ($epoch) {
 795         if (_bigfloatgt($other->dirtymark, $self->dirtymark||0)) {
 796             $oldest_allowed = 0;
 797             $something_done = 1;
 798         } elsif (my $merged = $self->merged) {
 799             my $secs = $self->interval_secs();
 800             $oldest_allowed = min($epoch - $secs, $merged->{epoch}||0);
 801             if (@$other_recent and
 802                 _bigfloatlt($other_recent->[-1]{epoch}, $oldest_allowed)
 803                ) {
 804                 $oldest_allowed = $other_recent->[-1]{epoch};
 805             }
 806         }
 807         while (@$my_recent && _bigfloatlt($my_recent->[-1]{epoch}, $oldest_allowed)) {
 808             pop @$my_recent;
 809             $something_done = 1;
 810         }
 811     }
 812
 813     my %have_path;
 814     my $other_recent_filtered = [];
 815     for my $oev (@$other_recent) {
 816         my $oevepoch = $oev->{epoch} || 0;
 817         next if _bigfloatlt($oevepoch, $oldest_allowed);
 818         my $path = $oev->{path};
 819         next if $have_path{$path}++;
 820         if (    $self->interval eq "Z"
 821             and $oev->{type}     eq "delete") {
 822             # do nothing
 823         } else {
 824             if (!$myepoch || _bigfloatgt($oevepoch, $myepoch)) {
 825                 $something_done = 1;
 826             }
 827             push @$other_recent_filtered, { epoch => $oev->{epoch}, path => $path, type => $oev->{type} };
 828         }
 829     }
 830     if ($something_done) {
 831         $self->_merge_something_done ($other_recent_filtered, $my_recent, $other_recent, $other, \%have_path, $epoch);
 832     }
 833     $self->unlock;
 834     $other->unlock;
 835 }
 836
 837 sub _merge_something_done {
 838     my($self, $other_recent_filtered, $my_recent, $other_recent, $other, $have_path, $epoch) = @_;
 839     my $recent = [];
 840     my $epoch_conflict = 0;
 841     my $last_epoch;
 842  ZIP: while (@$other_recent_filtered || @$my_recent) {
 843         my $event;
 844         if (!@$my_recent ||
 845             @$other_recent_filtered && _bigfloatge($other_recent_filtered->[0]{epoch},$my_recent->[0]{epoch})) {
 846             $event = shift @$other_recent_filtered;
 847         } else {
 848             $event = shift @$my_recent;
 849             next ZIP if $have_path->{$event->{path}}++;
 850         }
 851         $epoch_conflict=1 if defined $last_epoch && $event->{epoch} eq $last_epoch;
 852         $last_epoch = $event->{epoch};
 853         push @$recent, $event;
 854     }
 855     if ($epoch_conflict) {
 856         my %have_epoch;
 857         for (my $i = $#$recent;$i>=0;$i--) {
 858             my $epoch = $recent->[$i]{epoch};
 859             if ($have_epoch{$epoch}++) {
 860                 while ($have_epoch{$epoch}) {
 861                     $epoch = _increase_a_bit($epoch);
 862                 }
 863                 $recent->[$i]{epoch} = $epoch;
 864                 $have_epoch{$epoch}++;
 865             }
 866         }
 867     }
 868     if (!$self->dirtymark || _bigfloatgt($other->dirtymark, $self->dirtymark)) {
 869         $self->dirtymark ( $other->dirtymark );
 870     }
 871     $self->write_recent($recent);
 872     $other->merged({
 873                     time => Time::HiRes::time, # not used anywhere
 874                     epoch => $recent->[0]{epoch},
 875                     into_interval => $self->interval, # not used anywhere
 876                    });
 877     $other->write_recent($other_recent);
 878 }
 879
 880 sub _merge_sanitycheck {
 881     my($self, $other) = @_;
 882     if ($self->interval_secs <= $other->interval_secs) {
 883         die sprintf
 884             (
 885              "Alert: illegal merge operation of a bigger interval[%d] into a smaller[%d]",
 886              $self->interval_secs,
 887              $other->interval_secs,
 888             );
 889     }
 890 }
 891
 892 =head2 merged
 893
 894 Hashref denoting when this recentfile has been merged into some other
 895 at which epoch.
 896
 897 =cut
 898
 899 sub merged {
 900     my($self, $set) = @_;
 901     if (defined $set) {
 902         $self->_merged ($set);
 903     }
 904     my $merged = $self->_merged;
 905     my $into;
 906     if ($merged and $into = $merged->{into_interval} and defined $self->_interval) {
 907         # sanity checks
 908         if ($into eq $self->interval) {
 909             require Carp;
 910             Carp::cluck(sprintf
 911                 (
 912                  "Warning: into_interval[%s] same as own interval[%s]. Danger ahead.",
 913                  $into,
 914                  $self->interval,
 915                 ));
 916         } elsif ($self->interval_secs($into) < $self->interval_secs) {
 917             require Carp;
 918             Carp::cluck(sprintf
 919                 (
 920                  "Warning: into_interval_secs[%s] smaller than own interval_secs[%s] on interval[%s]. Danger ahead.",
 921                  $self->interval_secs($into),
 922                  $self->interval_secs,
 923                  $self->interval,
 924                 ));
 925         }
 926     }
 927     $merged;
 928 }
 929
 930 =head2 $hashref = $obj->meta_data
 931
 932 Returns the hashref of metadata that the server has to add to the
 933 I<recentfile>.
 934
 935 =cut
 936
 937 sub meta_data {
 938     my($self) = @_;
 939     my $ret = $self->{meta};
 940     for my $m (
 941                "aggregator",
 942                "canonize",
 943                "comment",
 944                "dirtymark",
 945                "filenameroot",
 946                "merged",
 947                "interval",
 948                "protocol",
 949                "serializer_suffix",
 950               ) {
 951         my $v = $self->$m;
 952         if (defined $v) {
 953             $ret->{$m} = $v;
 954         }
 955     }
 956     # XXX need to reset the Producer if I am a writer, keep it when I
 957     # am a reader
 958     $ret->{Producers} ||= {
 959                            __PACKAGE__, "$VERSION", # stringified it looks better
 960                            '$0', $0,
 961                            'time', Time::HiRes::time,
 962                           };
 963     $ret->{dirtymark} ||= Time::HiRes::time;
 964     return $ret;
 965 }
 966
 967 =head2 $success = $obj->mirror ( %options )
 968
 969 Mirrors the files in this I<recentfile> as reported by
 970 C<recent_events>. Options named C<after>, C<before>, C<max>, and
 971 C<skip-deletes> are passed through to the L<recent_events> call. The
 972 boolean option C<piecemeal>, if true, causes C<mirror> to only rsync
 973 C<max_files_per_connection> and keep track of the rsynced files so
 974 that future calls will rsync different files until all files are
 975 brought to sync.
 976
 977 =cut
 978
 979 sub mirror {
 980     my($self, %options) = @_;
 981     my $trecentfile = $self->get_remote_recentfile_as_tempfile();
 982     $self->_use_tempfile (1);
 983     my %passthrough = map { ($_ => $options{$_}) } qw(before after max skip-deletes);
 984     my ($recent_events) = $self->recent_events(%passthrough);
 985     my(@error, @xcollector);
 986     my $first_item = 0;
 987     my $last_item = $#$recent_events;
 988     my $done = $self->done;
 989     my $pathdb = $self->_pathdb;
 990   ITEM: for my $i ($first_item..$last_item) {
 991         my $status = +{};
 992         $self->_mirror_item
 993             (
 994              $i,
 995              $recent_events,
 996              $last_item,
 997              $done,
 998              $pathdb,
 999              \@xcollector,
1000              \%options,
1001              $status,
1002              \@error,
1003             );
1004         last if $i == $last_item;
1005         return if $status->{mustreturn};
1006     }
1007     if (@xcollector) {
1008         my $success = eval { $self->_mirror_empty_xcollector (\@xcollector,$pathdb,$recent_events);};
1009         if (!$success || $@) {
1010             warn "Warning: Unknown error while mirroring: $@";
1011             push @error, $@;
1012             sleep 1;
1013         }
1014     }
1015     if ($self->verbose) {
1016         print STDERR "DONE\n";
1017     }
1018     # once we've gone to the end we consider ourselves free of obligations
1019     $self->unseed;
1020     $self->_mirror_unhide_tempfile ($trecentfile);
1021     $self->_mirror_perform_delayed_ops;
1022     return !@error;
1023 }
1024
1025 sub _mirror_item {
1026     my($self,
1027        $i,
1028        $recent_events,
1029        $last_item,
1030        $done,
1031        $pathdb,
1032        $xcollector,
1033        $options,
1034        $status,
1035        $error,
1036       ) = @_;
1037     my $recent_event = $recent_events->[$i];
1038     return if $done->covered ( $recent_event->{epoch} );
1039     if ($pathdb) {
1040         my $rec = $pathdb->{$recent_event->{path}};
1041         if ($rec && $rec->{recentepoch}) {
1042             if (_bigfloatgt
1043                 ( $rec->{recentepoch}, $recent_event->{epoch} )){
1044                 $done->register ($recent_events, [$i]);
1045                 return;
1046             }
1047         }
1048     }
1049     my $dst = $self->local_path($recent_event->{path});
1050     if ($recent_event->{type} eq "new"){
1051         $self->_mirror_item_new
1052             (
1053              $dst,
1054              $i,
1055              $last_item,
1056              $recent_events,
1057              $recent_event,
1058              $xcollector,
1059              $pathdb,
1060              $status,
1061              $error,
1062              $options,
1063             );
1064     } elsif ($recent_event->{type} eq "delete") {
1065         my $activity;
1066         if ($options->{'skip-deletes'}) {
1067             $activity = "skipped";
1068         } else {
1069             if (! -e $dst) {
1070                 $activity = "not_found";
1071             } elsif (-l $dst or not -d _) {
1072                 $self->delayed_operations->{unlink}{$dst}++;
1073                 $activity = "deleted";
1074             } else {
1075                 $self->delayed_operations->{rmdir}{$dst}++;
1076                 $activity = "deleted";
1077             }
1078         }
1079         $done->register ($recent_events, [$i]);
1080         if ($pathdb) {
1081             $self->_mirror_register_path($pathdb,[$recent_event],$activity);
1082         }
1083     } else {
1084         warn "Warning: invalid upload type '$recent_event->{type}'";
1085     }
1086 }
1087
1088 sub _mirror_item_new {
1089     my($self,
1090        $dst,
1091        $i,
1092        $last_item,
1093        $recent_events,
1094        $recent_event,
1095        $xcollector,
1096        $pathdb,
1097        $status,
1098        $error,
1099        $options,
1100       ) = @_;
1101     if ($self->verbose) {
1102         my $doing = -e $dst ? "Sync" : "Get";
1103         printf STDERR
1104             (
1105              "%-4s %d (%d/%d/%s) %s ... ",
1106              $doing,
1107              time,
1108              1+$i,
1109              1+$last_item,
1110              $self->interval,
1111              $recent_event->{path},
1112             );
1113     }
1114     my $max_files_per_connection = $self->max_files_per_connection || 42;
1115     my $success;
1116     if ($self->verbose) {
1117         print STDERR "\n";
1118     }
1119     push @$xcollector, { rev => $recent_event, i => $i };
1120     if (@$xcollector >= $max_files_per_connection) {
1121         $success = eval {$self->_mirror_empty_xcollector ($xcollector,$pathdb,$recent_events);};
1122         my $sleep = $self->sleep_per_connection;
1123         $sleep = 0.42 unless defined $sleep;
1124         Time::HiRes::sleep $sleep;
1125         if ($options->{piecemeal}) {
1126             $status->{mustreturn} = 1;
1127             return;
1128         }
1129     } else {
1130         return;
1131     }
1132     if (!$success || $@) {
1133         warn "Warning: Error while mirroring: $@";
1134         push @$error, $@;
1135         sleep 1;
1136     }
1137     if ($self->verbose) {
1138         print STDERR "DONE\n";
1139     }
1140 }
1141
1142 sub _mirror_empty_xcollector {
1143     my($self,$xcoll,$pathdb,$recent_events) = @_;
1144     my $success = $self->mirror_path([map {$_->{rev}{path}} @$xcoll]);
1145     if ($pathdb) {
1146         $self->_mirror_register_path($pathdb,[map {$_->{rev}} @$xcoll],"rsync");
1147     }
1148     $self->done->register($recent_events, [map {$_->{i}} @$xcoll]);
1149     @$xcoll = ();
1150     return $success;
1151 }
1152
1153 sub _mirror_register_path {
1154     my($self,$pathdb,$coll,$activity) = @_;
1155     my $time = time;
1156     for my $item (@$coll) {
1157         $pathdb->{$item->{path}} =
1158             {
1159              recentepoch => $item->{epoch},
1160              ($activity."_on") => $time,
1161             };
1162     }
1163 }
1164
1165 sub _mirror_unhide_tempfile {
1166     my($self, $trecentfile) = @_;
1167     my $rfile = $self->rfile;
1168     if (rename $trecentfile, $rfile) {
1169         # warn "DEBUG: renamed '$trecentfile' to '$rfile'";
1170     } else {
1171         require Carp;
1172         Carp::confess("Could not rename '$trecentfile' to '$rfile': $!");
1173     }
1174     $self->_use_tempfile (0);
1175     if (my $ctfh = $self->_current_tempfile_fh) {
1176         $ctfh->unlink_on_destroy (0);
1177         $self->_current_tempfile_fh (undef);
1178     }
1179 }
1180
1181 sub _mirror_perform_delayed_ops {
1182     my($self) = @_;
1183     my $delayed = $self->delayed_operations;
1184     for my $dst (keys %{$delayed->{unlink}}) {
1185         unless (unlink $dst) {
1186             require Carp;
1187             Carp::cluck ( "Warning: Error while unlinking '$dst': $!" );
1188         }
1189         delete $delayed->{unlink}{$dst};
1190     }
1191     for my $dst (keys %{$delayed->{rmdir}}) {
1192         unless (rmdir $dst) {
1193             require Carp;
1194             Carp::cluck ( "Warning: Error on rmdir '$dst': $!" );
1195         }
1196         delete $delayed->{rmdir}{$dst};
1197     }
1198 }
1199
1200 =head2 (void) $obj->mirror_loop
1201
1202 Run mirror in an endless loop. See the accessor C<loopinterval>. XXX
1203 What happens/should happen if we miss the interval during a single loop?
1204
1205 =cut
1206
1207 sub mirror_loop {
1208     my($self) = @_;
1209     my $iteration_start = time;
1210
1211     my $Signal = 0;
1212     $SIG{INT} = sub { $Signal++ };
1213     my $loopinterval = $self->loopinterval || 42;
1214     my $after = -999999999;
1215   LOOP: while () {
1216         $self->mirror($after);
1217         last LOOP if $Signal;
1218         my $re = $self->recent_events;
1219         $after = $re->[0]{epoch};
1220         if ($self->verbose) {
1221             local $| = 1;
1222             print "($after)";
1223         }
1224         if (time - $iteration_start < $loopinterval) {
1225             sleep $iteration_start + $loopinterval - time;
1226         }
1227         if ($self->verbose) {
1228             local $| = 1;
1229             print "~";
1230         }
1231     }
1232 }
1233
1234 =head2 $success = $obj->mirror_path ( $arrref | $path )
1235
1236 If the argument is a scalar it is treated as a path. The remote path
1237 is mirrored into the local copy. $path is the path found in the
1238 I<recentfile>, i.e. it is relative to the root directory of the
1239 mirror.
1240
1241 If the argument is an array reference then all elements are treated as
1242 a path below the current tree and all are rsynced with a single
1243 command (and a single connection).
1244
1245 =cut
1246
1247 sub mirror_path {
1248     my($self,$path) = @_;
1249     # XXX simplify the two branches such that $path is treated as
1250     # [$path] maybe even demand the argument as an arrayref to
1251     # simplify docs and code. (rsync-over-recentfile-2.pl uses the
1252     # interface)
1253     if (ref $path and ref $path eq "ARRAY") {
1254         my $dst = $self->localroot;
1255         mkpath dirname $dst;
1256         my($fh) = File::Temp->new(TEMPLATE => sprintf(".%s-XXXX",
1257                                                       lc $self->filenameroot,
1258                                                      ),
1259                                   TMPDIR => 1,
1260                                   UNLINK => 0,
1261                                  );
1262         for my $p (@$path) {
1263             print $fh $p, "\n";
1264         }
1265         $fh->flush;
1266         $fh->unlink_on_destroy(1);
1267         my $gaveup = 0;
1268         my $retried = 0;
1269         while (!$self->rsync->exec
1270                (
1271                 src => join("/",
1272                             $self->remoteroot,
1273                            ),
1274                 dst => $dst,
1275                 'files-from' => $fh->filename,
1276                )) {
1277             my(@err) = $self->rsync->err;
1278             if ($self->ignore_link_stat_errors && "@err" =~ m{^ rsync: \s link_stat }x ) {
1279                 if ($self->verbose) {
1280                     warn "Info: ignoring link_stat error '@err'";
1281                 }
1282                 return 1;
1283             }
1284             $self->register_rsync_error (@err);
1285             if (++$retried >= 3) {
1286                 warn "XXX giving up.";
1287                 $gaveup = 1;
1288                 last;
1289             }
1290         }
1291         unless ($gaveup) {
1292             $self->un_register_rsync_error ();
1293         }
1294     } else {
1295         my $dst = $self->local_path($path);
1296         mkpath dirname $dst;
1297         while (!$self->rsync->exec
1298                (
1299                 src => join("/",
1300                             $self->remoteroot,
1301                             $path
1302                            ),
1303                 dst => $dst,
1304                 )) {
1305             my(@err) = $self->rsync->err;
1306             if ($self->ignore_link_stat_errors && "@err" =~ m{^ rsync: \s link_stat }x ) {
1307                 if ($self->verbose) {
1308                     warn "Info: ignoring link_stat error '@err'";
1309                 }
1310                 return 1;
1311             }
1312             $self->register_rsync_error (@err);
1313         }
1314         $self->un_register_rsync_error ();
1315     }
1316     return 1;
1317 }
1318
1319 sub _my_current_rfile {
1320     my($self) = @_;
1321     my $rfile;
1322     if ($self->_use_tempfile) {
1323         $rfile = $self->_current_tempfile;
1324     } else {
1325         $rfile = $self->rfile;
1326     }
1327     return $rfile;
1328 }
1329
1330 =head2 $path = $obj->naive_path_normalize ($path)
1331
1332 Takes an absolute unix style path as argument and canonicalizes it to
1333 a shorter path if possible, removing things like double slashes or
1334 C</./> and removes references to C<../> directories to get a shorter
1335 unambiguos path. This is used to make the code easier that determines
1336 if a file passed to C<upgrade()> is indeed below our C<localroot>.
1337
1338 =cut
1339
1340 sub naive_path_normalize {
1341     my($self,$path) = @_;
1342     $path =~ s|/+|/|g;
1343     1 while $path =~ s|/[^/]+/\.\./|/|;
1344     $path =~ s|/$||;
1345     $path;
1346 }
1347
1348 =head2 $ret = $obj->read_recent_1 ( $data )
1349
1350 Delegate of C<recent_events()> on protocol 1
1351
1352 =cut
1353
1354 sub read_recent_1 {
1355     my($self, $data) = @_;
1356     return $data->{recent};
1357 }
1358
1359 =head2 $array_ref = $obj->recent_events ( %options )
1360
1361 Note: the code relies on the resource being written atomically. We
1362 cannot lock because we may have no write access. If the caller has
1363 write access (eg. aggregate() or update()), it has to care for any
1364 necessary locking and it MUST write atomically.
1365
1366 If $options{after} is specified, only file events after this timestamp
1367 are returned.
1368
1369 If $options{before} is specified, only file events before this
1370 timestamp are returned.
1371
1372 IF $options{'skip-deletes'} is specified, no files-to-be-deleted will
1373 be returned.
1374
1375 If $options{max} is specified only a maximum of this many events is
1376 returned.
1377
1378 If $options{contains} is specified the value must be a hash reference
1379 containing a query. The query may contain the keys C<epoch>, C<path>,
1380 and C<type>. Each represents a condition that must be met. If there is
1381 more than one such key, the conditions are ANDed.
1382
1383 If $options{info} is specified, it must be a hashref. This hashref
1384 will be filled with metadata about the unfiltered recent_events of
1385 this object, in key C<first> there is the first item, in key C<last>
1386 is the last.
1387
1388 =cut
1389
1390 sub recent_events {
1391     my ($self, %options) = @_;
1392     my $info = $options{info};
1393     if ($self->is_slave) {
1394         $self->get_remote_recentfile_as_tempfile;
1395     }
1396     my $rfile_or_tempfile = $self->_my_current_rfile or return [];
1397     -e $rfile_or_tempfile or return [];
1398     my $suffix = $self->serializer_suffix;
1399     my ($data) = eval {
1400         $self->_try_deserialize
1401             (
1402              $suffix,
1403              $rfile_or_tempfile,
1404             );
1405     };
1406     my $err = $@;
1407     if ($err or !$data) {
1408         return [];
1409     }
1410     my $re;
1411     if (reftype $data eq 'ARRAY') { # protocol 0
1412         $re = $data;
1413     } else {
1414         $re = $self->_recent_events_protocol_x
1415             (
1416              $data,
1417              $rfile_or_tempfile,
1418             );
1419     }
1420     return $re unless grep {defined $options{$_}} qw(after before max);
1421     $self->_recent_events_handle_options ($re, \%options);
1422 }
1423
1424 sub _recent_events_handle_options {
1425     my($self, $re, $options) = @_;
1426     my $last_item = $#$re;
1427     my $info = $options->{info};
1428     if ($info) {
1429         $info->{first} = $re->[0];
1430         $info->{last} = $re->[-1];
1431     }
1432     if (defined $options->{after}) {
1433         if ($re->[0]{epoch} > $options->{after}) {
1434             if (
1435                 my $f = first
1436                         {$re->[$_]{epoch} <= $options->{after}}
1437                         0..$#$re
1438                ) {
1439                 $last_item = $f-1;
1440             }
1441         } else {
1442             $last_item = -1;
1443         }
1444     }
1445     my $first_item = 0;
1446     if (defined $options->{before}) {
1447         if ($re->[0]{epoch} > $options->{before}) {
1448             if (
1449                 my $f = first
1450                         {$re->[$_]{epoch} < $options->{before}}
1451                         0..$last_item
1452                ) {
1453                 $first_item = $f;
1454             }
1455         } else {
1456             $first_item = 0;
1457         }
1458     }
1459     if (0 != $first_item || -1 != $last_item) {
1460         @$re = splice @$re, $first_item, 1+$last_item-$first_item;
1461     }
1462     if ($options->{'skip-deletes'}) {
1463         @$re = grep { $_->{type} ne "delete" } @$re;
1464     }
1465     if (my $contopt = $options->{contains}) {
1466         my $seen_allowed = 0;
1467         for my $allow (qw(epoch path type)) {
1468             if (exists $contopt->{$allow}) {
1469                 $seen_allowed++;
1470                 my $v = $contopt->{$allow};
1471                 @$re = grep { $_->{$allow} eq $v } @$re;
1472             }
1473         }
1474         if (keys %$contopt > $seen_allowed) {
1475             require Carp;
1476             Carp::confess
1477                     (sprintf "unknown query: %s", join ", ", %$contopt);
1478         }
1479     }
1480     if ($options->{max} && @$re > $options->{max}) {
1481         @$re = splice @$re, 0, $options->{max};
1482     }
1483     $re;
1484 }
1485
1486 sub _recent_events_protocol_x {
1487     my($self,
1488        $data,
1489        $rfile_or_tempfile,
1490       ) = @_;
1491     my $meth = sprintf "read_recent_%d", $data->{meta}{protocol};
1492     # we may be reading meta for the first time
1493     while (my($k,$v) = each %{$data->{meta}}) {
1494         next if $k ne lc $k; # "Producers"
1495         next if defined $self->$k;
1496         $self->$k($v);
1497     }
1498     my $re = $self->$meth ($data);
1499     my @stat = stat $rfile_or_tempfile or die "Cannot stat '$rfile_or_tempfile': $!";
1500     my $minmax = { mtime => $stat[9] };
1501     if (@$re) {
1502         $minmax->{min} = $re->[-1]{epoch};
1503         $minmax->{max} = $re->[0]{epoch};
1504     }
1505     $self->minmax ( $minmax );
1506     return $re;
1507 }
1508
1509 sub _try_deserialize {
1510     my($self,
1511        $suffix,
1512        $rfile_or_tempfile,
1513       ) = @_;
1514     if ($suffix eq ".yaml") {
1515         require YAML::Syck;
1516         YAML::Syck::LoadFile($rfile_or_tempfile);
1517     } elsif ($HAVE->{"Data::Serializer"}) {
1518         my $serializer = Data::Serializer->new
1519             ( serializer => $serializers{$suffix} );
1520         my $serialized = do
1521             {
1522                 open my $fh, $rfile_or_tempfile or die "Could not open: $!";
1523                 local $/;
1524                 <$fh>;
1525             };
1526         $serializer->raw_deserialize($serialized);
1527     } else {
1528         die "Data::Serializer not installed, cannot proceed with suffix '$suffix'";
1529     }
1530 }
1531
1532 sub _refresh_internals {
1533     my($self, $dst) = @_;
1534     my $class = ref $self;
1535     my $rfpeek = $class->new_from_file ($dst);
1536     for my $acc (qw(
1537                     _merged
1538                     minmax
1539                    )) {
1540         $self->$acc ( $rfpeek->$acc );
1541     }
1542     my $old_dirtymark = $self->dirtymark;
1543     my $new_dirtymark = $rfpeek->dirtymark;
1544     if ($old_dirtymark && $new_dirtymark && _bigfloatgt($new_dirtymark,$old_dirtymark)) {
1545         $self->done->reset;
1546         $self->dirtymark ( $new_dirtymark );
1547         $self->seed;
1548     }
1549 }
1550
1551 =head2 $ret = $obj->rfilename
1552
1553 Just the basename of our I<recentfile>, composed from C<filenameroot>,
1554 a dash, C<interval>, and C<serializer_suffix>. E.g. C<RECENT-6h.yaml>
1555
1556 =cut
1557
1558 sub rfilename {
1559     my($self) = @_;
1560     my $file = sprintf("%s-%s%s",
1561                        $self->filenameroot,
1562                        $self->interval,
1563                        $self->serializer_suffix,
1564                       );
1565     return $file;
1566 }
1567
1568 =head2 $str = $self->remote_dir
1569
1570 The directory we are mirroring from.
1571
1572 =cut
1573
1574 sub remote_dir {
1575     my($self, $set) = @_;
1576     if (defined $set) {
1577         $self->_remote_dir ($set);
1578     }
1579     my $x = $self->_remote_dir;
1580     $self->is_slave (1);
1581     return $x;
1582 }
1583
1584 =head2 $str = $obj->remoteroot
1585
1586 =head2 (void) $obj->remoteroot ( $set )
1587
1588 Get/Set the composed prefix needed when rsyncing from a remote module.
1589 If remote_host, remote_module, and remote_dir are set, it is composed
1590 from these.
1591
1592 =cut
1593
1594 sub remoteroot {
1595     my($self, $set) = @_;
1596     if (defined $set) {
1597         $self->_remoteroot($set);
1598     }
1599     my $remoteroot = $self->_remoteroot;
1600     unless (defined $remoteroot) {
1601         $remoteroot = sprintf
1602             (
1603              "%s%s%s",
1604              defined $self->remote_host   ? ($self->remote_host."::")  : "",
1605              defined $self->remote_module ? ($self->remote_module."/") : "",
1606              defined $self->remote_dir    ? $self->remote_dir          : "",
1607             );
1608         $self->_remoteroot($remoteroot);
1609     }
1610     return $remoteroot;
1611 }
1612
1613 =head2 (void) $obj->resolve_recentfilename ( $recentfilename )
1614
1615 Inverse method to L<rfilename>. $recentfilename is a plain filename of
1616 the pattern
1617
1618     $filenameroot-$interval$serializer_suffix
1619
1620 e.g.
1621
1622     RECENT-1M.yaml
1623
1624 This filename is split into its parts and the parts are fed to the
1625 object itself.
1626
1627 =cut
1628
1629 sub resolve_recentfilename {
1630     my($self, $rfname) = @_;
1631     my($splitter) = qr(^(.+)-([^-\.]+)(\.[^\.]+));
1632     if (my($f,$i,$s) = $rfname =~ $splitter) {
1633         $self->filenameroot      ($f);
1634         $self->interval          ($i);
1635         $self->serializer_suffix ($s);
1636     } else {
1637         die "Alert: cannot split '$rfname', doesn't match '$splitter'";
1638     }
1639     return;
1640 }
1641
1642 =head2 my $rfile = $obj->rfile
1643
1644 Returns the full path of the I<recentfile>
1645
1646 =cut
1647
1648 sub rfile {
1649     my($self) = @_;
1650     my $rfile = $self->_rfile;
1651     return $rfile if defined $rfile;
1652     $rfile = File::Spec->catfile
1653         ($self->localroot,
1654          $self->rfilename,
1655         );
1656     $self->_rfile ($rfile);
1657     return $rfile;
1658 }
1659
1660 =head2 $rsync_obj = $obj->rsync
1661
1662 The File::Rsync object that this object uses for communicating with an
1663 upstream server.
1664
1665 =cut
1666
1667 sub rsync {
1668     my($self) = @_;
1669     my $rsync = $self->_rsync;
1670     unless (defined $rsync) {
1671         my $rsync_options = $self->rsync_options || {};
1672         if ($HAVE->{"File::Rsync"}) {
1673             $rsync = File::Rsync->new($rsync_options);
1674             $self->_rsync($rsync);
1675         } else {
1676             die "File::Rsync required for rsync operations. Cannot continue";
1677         }
1678     }
1679     return $rsync;
1680 }
1681
1682 =head2 (void) $obj->register_rsync_error(@err)
1683
1684 =head2 (void) $obj->un_register_rsync_error()
1685
1686 Register_rsync_error is called whenever the File::Rsync object fails
1687 on an exec (say, connection doesn't succeed). It issues a warning and
1688 sleeps for an increasing amount of time. Un_register_rsync_error
1689 resets the error count. See also accessor C<max_rsync_errors>.
1690
1691 =cut
1692
1693 {
1694     my $no_success_count = 0;
1695     my $no_success_time = 0;
1696     sub register_rsync_error {
1697         my($self, @err) = @_;
1698         chomp @err;
1699         $no_success_time = time;
1700         $no_success_count++;
1701         my $max_rsync_errors = $self->max_rsync_errors;
1702         $max_rsync_errors = MAX_INT unless defined $max_rsync_errors;
1703         if ($max_rsync_errors>=0 && $no_success_count >= $max_rsync_errors) {
1704             require Carp;
1705             Carp::confess
1706                   (
1707                    sprintf
1708                    (
1709                     "Alert: Error while rsyncing (%s): '%s', error count: %d, exiting now,",
1710                     $self->interval,
1711                     join(" ",@err),
1712                     $no_success_count,
1713                    ));
1714         }
1715         my $sleep = 12 * $no_success_count;
1716         $sleep = 300 if $sleep > 300;
1717         require Carp;
1718         Carp::cluck
1719               (sprintf
1720                (
1721                 "Warning: %s, Error while rsyncing (%s): '%s', sleeping %d",
1722                 scalar(localtime($no_success_time)),
1723                 $self->interval,
1724                 join(" ",@err),
1725                 $sleep,
1726                ));
1727         sleep $sleep
1728     }
1729     sub un_register_rsync_error {
1730         my($self) = @_;
1731         $no_success_time = 0;
1732         $no_success_count = 0;
1733     }
1734 }
1735
1736 =head2 $clone = $obj->_sparse_clone
1737
1738 Clones just as much from itself that it does not hurt. Experimental
1739 method.
1740
1741 Note: what fits better: sparse or shallow? Other suggestions?
1742
1743 =cut
1744
1745 sub _sparse_clone {
1746     my($self) = @_;
1747     my $new = bless {}, ref $self;
1748     for my $m (qw(
1749                   _interval
1750                   _localroot
1751                   _remoteroot
1752                   _rfile
1753                   _use_tempfile
1754                   aggregator
1755                   filenameroot
1756                   is_slave
1757                   max_files_per_connection
1758                   protocol
1759                   rsync_options
1760                   serializer_suffix
1761                   sleep_per_connection
1762                   verbose
1763                  )) {
1764         my $o = $self->$m;
1765         $o = Storable::dclone $o if ref $o;
1766         $new->$m($o);
1767     }
1768     $new;
1769 }
1770
1771 =head2 $boolean = OBJ->ttl_reached ()
1772
1773 =cut
1774
1775 sub ttl_reached {
1776     my($self) = @_;
1777     my $have_mirrored = $self->have_mirrored || 0;
1778     my $now = Time::HiRes::time;
1779     my $ttl = $self->ttl;
1780     $ttl = 24.2 unless defined $ttl;
1781     if ($now > $have_mirrored + $ttl) {
1782         return 1;
1783     }
1784     return 0;
1785 }
1786
1787 =head2 (void) $obj->unlock()
1788
1789 Unlocking is implemented with an C<rmdir> on a locking directory
1790 (C<.lock> appended to $rfile).
1791
1792 =cut
1793
1794 sub unlock {
1795     my($self) = @_;
1796     return unless $self->_is_locked;
1797     my $rfile = $self->rfile;
1798     rmdir "$rfile.lock";
1799     $self->_is_locked (0);
1800 }
1801
1802 =head2 unseed
1803
1804 Sets this recentfile in the state of not 'seeded'.
1805
1806 =cut
1807 sub unseed {
1808     my($self) = @_;
1809     $self->seeded(0);
1810 }
1811
1812 =head2 $ret = $obj->update ($path, $type)
1813
1814 =head2 $ret = $obj->update ($path, "new", $dirty_epoch)
1815
1816 =head2 $ret = $obj->update ()
1817
1818 Enter one file into the local I<recentfile>. $path is the (usually
1819 absolute) path. If the path is outside I<our> tree, then it is
1820 ignored.
1821
1822 $type is one of C<new> or C<delete>.
1823
1824 Events of type C<new> may set $dirty_epoch. $dirty_epoch is normally
1825 not used and the epoch is calculated by the update() routine itself
1826 based on current time. But if there is the demand to insert a
1827 not-so-current file into the dataset, then the caller sets
1828 $dirty_epoch. This causes the epoch of the registered event to become
1829 $dirty_epoch or -- if the exact value given is already taken -- a tiny
1830 bit more. As compensation the dirtymark of the whole dataset is set to
1831 the current epoch.
1832
1833 The new file event is unshifted (or, if dirty_epoch is set, inserted
1834 at the place it belongs to, according to the rule to have a sequence
1835 of strictly decreasing timestamps) to the array of recent_events and
1836 the array is shortened to the length of the timespan allowed. This is
1837 usually the timespan specified by the interval of this recentfile but
1838 as long as this recentfile has not been merged to another one, the
1839 timespan may grow without bounds.
1840
1841 The third form runs an update without inserting a new file. This may
1842 be disired to truncate a recentfile.
1843
1844 =cut
1845 sub _epoch_monotonically_increasing {
1846     my($self,$epoch,$recent) = @_;
1847     return $epoch unless @$recent; # the first one goes unoffended
1848     if (_bigfloatgt("".$epoch,$recent->[0]{epoch})) {
1849         return $epoch;
1850     } else {
1851         return _increase_a_bit($recent->[0]{epoch});
1852     }
1853 }
1854 sub update {
1855     my($self,$path,$type,$dirty_epoch) = @_;
1856     if (defined $path or defined $type or defined $dirty_epoch) {
1857         die "update called without path argument" unless defined $path;
1858         die "update called without type argument" unless defined $type;
1859         die "update called with illegal type argument: $type" unless $type =~ /(new|delete)/;
1860         die "update called with \$type=$type and \$dirty_epoch=$dirty_epoch; ".
1861             "dirty_epoch only allowed with type=new" if $dirty_epoch and $type ne "new";
1862         my $canonmeth = $self->canonize;
1863         unless ($canonmeth) {
1864             $canonmeth = "naive_path_normalize";
1865         }
1866         $path = $self->$canonmeth($path);
1867     }
1868     my $lrd = $self->localroot;
1869     $self->lock;
1870     # you must calculate the time after having locked, of course
1871     my $now = Time::HiRes::time;
1872     my $interval = $self->interval;
1873     my $secs = $self->interval_secs();
1874     my $recent = $self->recent_events;
1875
1876     my $epoch;
1877     if ($dirty_epoch) {
1878         $epoch = $dirty_epoch;
1879     } else {
1880         $epoch = $self->_epoch_monotonically_increasing($now,$recent);
1881     }
1882
1883     $recent ||= [];
1884     my $oldest_allowed = 0;
1885     my $merged = $self->merged;
1886     if ($merged->{epoch}) {
1887         my $virtualnow = max($now,$epoch);
1888         # for the lower bound could we need big math?
1889         $oldest_allowed = min($virtualnow - $secs, $merged->{epoch}, $epoch);
1890     } else {
1891         # as long as we are not merged at all, no limits!
1892     }
1893     my $something_done = 0;
1894  TRUNCATE: while (@$recent) {
1895         # $DB::single++ unless defined $oldest_allowed;
1896         if (_bigfloatlt($recent->[-1]{epoch}, $oldest_allowed)) {
1897             pop @$recent;
1898             $something_done = 1;
1899         } else {
1900             last TRUNCATE;
1901         }
1902     }
1903     if (defined $path && $path =~ s|^\Q$lrd\E||) {
1904         $path =~ s|^/||;
1905         my $splicepos;
1906         # remove the older duplicates of this $path, irrespective of $type:
1907         if ($dirty_epoch) {
1908             my $ctx = $self->_update_with_dirty_epoch($path,$recent,$epoch);
1909             $recent    = $ctx->{recent};
1910             $splicepos = $ctx->{splicepos};
1911             $epoch     = $ctx->{epoch};
1912             my $dirtymark = $self->dirtymark;
1913             my $new_dm = $now;
1914             if (_bigfloatgt($epoch, $now)) {
1915                 $new_dm = $epoch;
1916             }
1917             $self->dirtymark($new_dm);
1918             my $merged = $self->merged;
1919             if (not defined $merged->{epoch} or _bigfloatlt($epoch,$merged->{epoch})) {
1920                 $self->merged(+{});
1921             }
1922         } else {
1923             $recent = [ grep { $_->{path} ne $path } @$recent ];
1924             $splicepos = 0;
1925         }
1926         if (defined $splicepos) {
1927             splice @$recent, $splicepos, 0, { epoch => $epoch, path => $path, type => $type };
1928         }
1929         $something_done = 1;
1930     }
1931
1932     $self->write_recent($recent) if $something_done;
1933     $self->_assert_symlink;
1934     $self->unlock;
1935 }
1936
1937 sub _update_with_dirty_epoch {
1938     my($self,$path,$recent,$epoch) = @_;
1939     my $splicepos;
1940     my $new_recent = [];
1941     if (grep { $_->{path} ne $path } @$recent) {
1942         my $cancel = 0;
1943     KNOWN_EVENT: for my $i (0..$#$recent) {
1944             if ($recent->[$i]{path} eq $path) {
1945                 if ($recent->[$i]{epoch} eq $epoch) {
1946                     # nothing to do
1947                     $cancel = 1;
1948                     last KNOWN_EVENT;
1949                 }
1950             } else {
1951                 push @$new_recent, $recent->[$i];
1952             }
1953         }
1954         @$recent = @$new_recent unless $cancel;
1955     }
1956     if (!exists $recent->[0] or _bigfloatgt($epoch,$recent->[0]{epoch})) {
1957         $splicepos = 0;
1958     } elsif (_bigfloatlt($epoch,$recent->[0]{epoch})) {
1959         $splicepos = @$recent;
1960     } else {
1961     RECENT: for my $i (0..$#$recent) {
1962             my $ev = $recent->[$i];
1963             if ($epoch eq $recent->[$i]{epoch}) {
1964                 $epoch = _increase_a_bit($epoch, $i ? $recent->[$i-1]{epoch} : undef);
1965             }
1966             if (_bigfloatgt($epoch,$recent->[$i]{epoch})) {
1967                 $splicepos = $i;
1968                 last RECENT;
1969             }
1970         }
1971     }
1972     return {
1973             recent => $recent,
1974             splicepos => $splicepos,
1975             epoch => $epoch,
1976     }
1977 }
1978
1979 =head2 seed
1980
1981 Sets this recentfile in the state of 'seeded' which means it has to
1982 re-evaluate its uptodateness.
1983
1984 =cut
1985 sub seed {
1986     my($self) = @_;
1987     $self->seeded(1);
1988 }
1989
1990 =head2 seeded
1991
1992 Tells if the recentfile is in the state 'seeded'.
1993
1994 =cut
1995 sub seeded {
1996     my($self, $set) = @_;
1997     if (defined $set) {
1998         $self->_seeded ($set);
1999     }
2000     my $x = $self->_seeded;
2001     unless (defined $x) {
2002         $x = 0;
2003         $self->_seeded ($x);
2004     }
2005     return $x;
2006 }
2007
2008 =head2 uptodate
2009
2010 True if this object has mirrored the complete interval covered by the
2011 current recentfile.
2012
2013 *** WIP ***
2014
2015 =cut
2016 sub uptodate {
2017     my($self) = @_;
2018     my $uptodate;
2019     my $why;
2020     if ($self->_uptodateness_ever_reached and not $self->seeded) {
2021         $why = "saturated";
2022         $uptodate = 1;
2023     }
2024     unless (defined $uptodate) {
2025         if ($self->ttl_reached){
2026             $why = "ttl_reached returned true, so we are not uptodate";
2027             $uptodate = 0 ;
2028         }
2029     }
2030     unless (defined $uptodate) {
2031         # look if recentfile has unchanged timestamp
2032         my $minmax = $self->minmax;
2033         if (exists $minmax->{mtime}) {
2034             my $rfile = $self->_my_current_rfile;
2035             my @stat = stat $rfile;
2036             my $mtime = $stat[9];
2037             if ($mtime > $minmax->{mtime}) {
2038                 $why = "mtime[$mtime] of rfile[$rfile] > minmax/mtime[$minmax->{mtime}], so we are not uptodate";
2039                 $uptodate = 0;
2040             } else {
2041                 my $covered = $self->done->covered(@$minmax{qw(max min)});
2042                 $why = "minmax covered[$covered], so we return that";
2043                 $uptodate = $covered;
2044             }
2045         }
2046     }
2047     unless (defined $uptodate) {
2048         $why = "fallthrough, so not uptodate";
2049         $uptodate = 0;
2050     }
2051     if ($uptodate) {
2052         $self->_uptodateness_ever_reached(1);
2053         $self->unseed;
2054     }
2055     my $remember =
2056         {
2057          uptodate => $uptodate,
2058          why => $why,
2059         };
2060     $self->_remember_last_uptodate_call($remember);
2061     return $uptodate;
2062 }
2063
2064 =head2 $obj->write_recent ($recent_files_arrayref)
2065
2066 Writes a I<recentfile> based on the current reflection of the current
2067 state of the tree limited by the current interval.
2068
2069 =cut
2070 sub _resort {
2071     my($self,$recent) = @_;
2072     @$recent = sort { _bigfloatcmp($b->{epoch},$a->{epoch}) } @$recent;
2073     return;
2074 }
2075 sub write_recent {
2076     my ($self,$recent) = @_;
2077     die "write_recent called without argument" unless defined $recent;
2078     my $Last_epoch;
2079  SANITYCHECK: for my $i (0..$#$recent) {
2080         if (defined $Last_epoch && _bigfloatge($recent->[$i]{epoch},$Last_epoch)) {
2081             warn sprintf "Warning: not-monotonic sequence '$recent->[$i]{epoch}'>='$Last_epoch', resorting %s\n", $self->interval;
2082             $self->_resort($recent);
2083             last SANITYCHECK;
2084         }
2085         $Last_epoch = $recent->[$i]{epoch};
2086     }
2087     my $meth = sprintf "write_%d", $self->protocol;
2088     $self->$meth($recent);
2089 }
2090
2091 =head2 $obj->write_0 ($recent_files_arrayref)
2092
2093 Delegate of C<write_recent()> on protocol 0
2094
2095 =cut
2096
2097 sub write_0 {
2098     my ($self,$recent) = @_;
2099     my $rfile = $self->rfile;
2100     YAML::Syck::DumpFile("$rfile.new",$recent);
2101     rename "$rfile.new", $rfile or die "Could not rename to '$rfile': $!";
2102 }
2103
2104 =head2 $obj->write_1 ($recent_files_arrayref)
2105
2106 Delegate of C<write_recent()> on protocol 1
2107
2108 =cut
2109
2110 sub write_1 {
2111     my ($self,$recent) = @_;
2112     my $rfile = $self->rfile;
2113     my $suffix = $self->serializer_suffix;
2114     my $data = {
2115                 meta => $self->meta_data,
2116                 recent => $recent,
2117                };
2118     my $serialized;
2119     if ($suffix eq ".yaml") {
2120         $serialized = YAML::Syck::Dump($data);
2121     } elsif ($HAVE->{"Data::Serializer"}) {
2122         my $serializer = Data::Serializer->new
2123             ( serializer => $serializers{$suffix} );
2124         $serialized = $serializer->raw_serialize($data);
2125     } else {
2126         die "Data::Serializer not installed, cannot proceed with suffix '$suffix'";
2127     }
2128     open my $fh, ">", "$rfile.new" or die "Could not open >'$rfile.new': $!";
2129     print $fh $serialized;
2130     close $fh or die "Could not close '$rfile.new': $!";
2131     rename "$rfile.new", $rfile or die "Could not rename to '$rfile': $!";
2132 }
2133
2134 BEGIN {
2135     my @pod_lines =
2136         split /\n/, <<'=cut'; %serializers = map { eval } grep {s/^=item\s+C<<(.+)>>$/$1/} @pod_lines; }
2137
2138 =head1 THE ARCHITECTURE OF A COLLECTION OF RECENTFILES
2139
2140 The idea is that we want to have a short file that records really
2141 recent changes. So that a fresh mirror can be kept fresh as long as
2142 the connectivity is given. Then we want longer files that record the
2143 history before. So when the mirror falls behind the update period
2144 reflected in the shortest file, it can complement the list of recent
2145 file events with the next one. And if this is not long enough we want
2146 another one, again a bit longer. And we want one that completes the
2147 history back to the oldest file. The index files do contain the
2148 complete list of current files. The longer a period covered by an
2149 index file is gone the less often the index file is updated. For
2150 practical reasons adjacent files will often overlap a bit but this is
2151 neither necessary nor enforced. That's the basic idea. The following
2152 example represents a tree that has a few updates every day:
2153
2154  RECENT.recent -> RECENT-1h.yaml
2155  RECENT-6h.yaml
2156  RECENT-1d.yaml
2157  RECENT-1M.yaml
2158  RECENT-1W.yaml
2159  RECENT-1Q.yaml
2160  RECENT-1Y.yaml
2161  RECENT-Z.yaml
2162
2163 The first file is the principal file, in so far it is the one that is
2164 written first after a filesystem change. Usually a symlink links to it
2165 with a filename that has the same filenameroot and the suffix
2166 C<.recent>. On systems that do not support symlinks there is a plain
2167 copy maintained instead.
2168
2169 The last file, the Z file, contains the complementary files that are
2170 in none of the other files. It does never contain C<deletes>. Besides
2171 this it serves the role of a recovery mechanism or spill over pond.
2172 When things go wrong, it's a valuable controlling instance to hold the
2173 differences between the collection of limited interval files and the
2174 actual filesystem.
2175
2176 =head2 THE INDIVIDUAL RECENTFILE
2177
2178 A I<recentfile> consists of a hash that has two keys: C<meta> and
2179 C<recent>. The C<meta> part has metadata and the C<recent> part has a
2180 list of fileobjects.
2181
2182 =head2 THE META PART
2183
2184 Here we find things that are pretty much self explaining: all
2185 lowercase attributes are accessors and as such explained somewhere
2186 above in this manpage. The uppercase attribute C<Producers> contains
2187 version information about involved software components. Nothing to
2188 worry about as I believe.
2189
2190 =head2 THE RECENT PART
2191
2192 This is the interesting part. Every entry refers to some filesystem
2193 change (with path, epoch, type). The epoch value is the point in time
2194 when some change was I<registered>. Do not be tempted to believe that
2195 the entry has a direct relation to something like modification time or
2196 change time on the filesystem level. The timestamp (I<epoch> element)
2197 is a floating point number and does practically never correspond
2198 exactly to the data recorded in the filesystem but rather to the time
2199 when some process succeeded to report some filesystem change to the
2200 I<recentfile> mechanism. This is why many parts of the code refer to
2201 I<events>, because we merely try to record the I<event> of the
2202 discovery of a change, not the time of the change itself.
2203
2204 All these entries can be devided into two types (denoted by the
2205 C<type> attribute): C<new>s and C<delete>s. Changes and creations are
2206 C<new>s. Deletes are C<delete>s.
2207
2208 Besides an C<epoch> and a C<type> attribute we find a third one:
2209 C<path>. This path is relative to the directory we find the
2210 I<recentfile> in.
2211
2212 The order of the entries in the I<recentfile> is by decreasing epoch
2213 attribute. These are unique floating point numbers. When the server
2214 has ntp running correctly, then the timestamps are usually reflecting
2215 a real epoch. If time is running backwards, we trump the system epoch
2216 with strictly monotonically increasing floating point timestamps and
2217 guarantee they are unique.
2218
2219 =head1 CORRUPTION AND RECOVERY
2220
2221 If the origin host breaks the promise to deliver consistent and
2222 complete I<recentfiles> then the way back to sanity shall be achieved
2223 through traditional rsyncing between the hosts. But don't forget to
2224 report it as a bug:)
2225
2226 =head1 SERIALIZERS
2227
2228 The following suffixes are supported and trigger the use of these
2229 serializers:
2230
2231 =over 4
2232
2233 =item C<< ".yaml" => "YAML::Syck" >>
2234
2235 =item C<< ".json" => "JSON" >>
2236
2237 =item C<< ".sto"  => "Storable" >>
2238
2239 =item C<< ".dd"   => "Data::Dumper" >>
2240
2241 =back
2242
2243 =cut
2244
2245 BEGIN {
2246     my @pod_lines =
2247         split /\n/, <<'=cut'; %seconds = map { eval } grep {s/^=item\s+C<<(.+)>>$/$1/} @pod_lines; }
2248
2249 =head1 INTERVAL SPEC
2250
2251 An interval spec is a primitive way to express time spans. Normally it
2252 is composed from an integer and a letter.
2253
2254 As a special case, a string that consists only of the single letter
2255 C<Z>, stands for unlimited time.
2256
2257 The following letters express the specified number of seconds:
2258
2259 =over 4
2260
2261 =item C<< s => 1 >>
2262
2263 =item C<< m => 60 >>
2264
2265 =item C<< h => 60*60 >>
2266
2267 =item C<< d => 60*60*24 >>
2268
2269 =item C<< W => 60*60*24*7 >>
2270
2271 =item C<< M => 60*60*24*30 >>
2272
2273 =item C<< Q => 60*60*24*90 >>
2274
2275 =item C<< Y => 60*60*24*365.25 >>
2276
2277 =back
2278
2279 =cut
2280
2281 =head1 BACKGROUND
2282
2283 This is about speeding up rsync operation on large trees. Uses a small
2284 metadata cocktail and pull technology.
2285
2286 =head2 NON-COMPETITORS
2287
2288  File::Mirror        JWU/File-Mirror/File-Mirror-0.10.tar.gz only local trees
2289  Mirror::YAML        ADAMK/Mirror-YAML-0.03.tar.gz           some sort of inner circle
2290  Net::DownloadMirror KNORR/Net-DownloadMirror-0.04.tar.gz    FTP sites and stuff
2291  Net::MirrorDir      KNORR/Net-MirrorDir-0.05.tar.gz         dito
2292  Net::UploadMirror   KNORR/Net-UploadMirror-0.06.tar.gz      dito
2293  Pushmi::Mirror      CLKAO/Pushmi-v1.0.0.tar.gz              something SVK
2294
2295  rsnapshot           www.rsnapshot.org                       focus on backup
2296  csync               www.csync.org                           more like unison
2297  multi-rsync         sourceforge 167893                      lan push to many
2298
2299 =head2 COMPETITORS
2300
2301 The problem to solve which clusters and ftp mirrors and otherwise
2302 replicated datasets like CPAN share: how to transfer only a minimum
2303 amount of data to determine the diff between two hosts.
2304
2305 Normally it takes a long time to determine the diff itself before it
2306 can be transferred. Known solutions at the time of this writing are
2307 csync2, and rsync 3 batch mode.
2308
2309 For many years the best solution was csync2 which solves the problem
2310 by maintaining a sqlite database on both ends and talking a highly
2311 sophisticated protocol to quickly determine which files to send and
2312 which to delete at any given point in time. Csync2 is often
2313 inconvenient because it is push technology and the act of syncing
2314 demands quite an intimate relationship between the sender and the
2315 receiver. This is hard to achieve in an environment of loosely coupled
2316 sites where the number of sites is large or connections are
2317 unreliable or network topology is changing.
2318
2319 Rsync 3 batch mode works around these problems by providing rsync-able
2320 batch files which allow receiving nodes to replay the history of the
2321 other nodes. This reduces the need to have an incestuous relation but
2322 it has the disadvantage that these batch files replicate the contents
2323 of the involved files. This seems inappropriate when the nodes already
2324 have a means of communicating over rsync.
2325
2326 rersyncrecent solves this problem with a couple of (usually 2-10)
2327 index files which cover different overlapping time intervals. The
2328 master writes these files and the clients/slaves can construct the
2329 full tree from the information contained in them. The most recent
2330 index file usually covers the last seconds or minutes or hours of the
2331 tree and depending on the needs, slaves can rsync every few seconds or
2332 minutes and then bring their trees in full sync.
2333
2334 The rersyncrecent mode was developed for CPAN but I hope it is a
2335 convenient and economic general purpose solution. I'm looking forward
2336 to see a CPAN backbone that is only a few seconds behind PAUSE. And
2337 then ... the first FUSE based CPAN filesystem anyone?
2338
2339 =head1 FUTURE DIRECTIONS
2340
2341 Currently the origin server must keep track of injected and removed
2342 files. Should be supported by an inotify-based assistant.
2343
2344 =head1 SEE ALSO
2345
2346 Barbie is providing a database of release dates. See
2347 http://use.perl.org/~barbie/journal/37907
2348
2349 =head1 AUTHOR
2350
2351 Andreas König
2352
2353 =head1 BUGS
2354
2355 Please report any bugs or feature requests through the web interface
2356 at
2357 L<http://rt.cpan.org/NoAuth/ReportBug.html?Queue=File-Rsync-Mirror-Recentfile>.
2358 I will be notified, and then you'll automatically be notified of
2359 progress on your bug as I make changes.
2360
2361 =head1 SUPPORT
2362
2363 You can find documentation for this module with the perldoc command.
2364
2365     perldoc File::Rsync::Mirror::Recentfile
2366
2367 You can also look for information at:
2368
2369 =over 4
2370
2371 =item * RT: CPAN's request tracker
2372
2373 L<http://rt.cpan.org/NoAuth/Bugs.html?Dist=File-Rsync-Mirror-Recentfile>
2374
2375 =item * AnnoCPAN: Annotated CPAN documentation
2376
2377 L<http://annocpan.org/dist/File-Rsync-Mirror-Recentfile>
2378
2379 =item * CPAN Ratings
2380
2381 L<http://cpanratings.perl.org/d/File-Rsync-Mirror-Recentfile>
2382
2383 =item * Search CPAN
2384
2385 L<http://search.cpan.org/dist/File-Rsync-Mirror-Recentfile>
2386
2387 =back
2388
2389
2390 =head1 ACKNOWLEDGEMENTS
2391
2392 Thanks to RJBS for module-starter.
2393
2394 =head1 COPYRIGHT & LICENSE
2395
2396 Copyright 2008,2009 Andreas König.
2397
2398 This program is free software; you can redistribute it and/or modify it
2399 under the same terms as Perl itself.
2400
2401
2402 =cut
2403
2404 1; # End of File::Rsync::Mirror::Recentfile
2405
2406 # Local Variables:
2407 # mode: cperl
2408 # cperl-indent-level: 4
2409 # End: