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