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