lib/PublicInbox/Emergency.pm

   1 # Copyright (C) 2016-2020 all contributors <meta@public-inbox.org>
   2 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
   3 #
   4 # Emergency Maildir delivery for MDA
   5 package PublicInbox::Emergency;
   6 use strict;
   7 use warnings;
   8 use Fcntl qw(:DEFAULT SEEK_SET);
   9 use Sys::Hostname qw(hostname);
  10 use IO::Handle; # ->flush, ->autoflush
  11
  12 sub new {
  13         my ($class, $dir) = @_;
  14
  15         -d $dir or mkdir($dir) or die "failed to mkdir($dir): $!\n";
  16         foreach (qw(new tmp cur)) {
  17                 my $d = "$dir/$_";
  18                 next if -d $d;
  19                 -d $d or mkdir($d) or die "failed to mkdir($d): $!\n";
  20         }
  21         bless { dir => $dir, files => {}, t => 0, cnt => 0, pid => $$ }, $class;
  22 }
  23
  24 sub _fn_in {
  25         my ($self, $dir) = @_;
  26         my @host = split(/\./, hostname);
  27         my $now = time;
  28         if ($self->{t} != $now) {
  29                 $self->{t} = $now;
  30                 $self->{cnt} = 0;
  31         } else {
  32                 $self->{cnt}++;
  33         }
  34
  35         my $f;
  36         do {
  37                 $f = "$self->{dir}/$dir/$self->{t}.$$"."_$self->{cnt}.$host[0]";
  38                 $self->{cnt}++;
  39         } while (-e $f);
  40         $f;
  41 }
  42
  43 sub prepare {
  44         my ($self, $strref) = @_;
  45
  46         die "already in transaction: $self->{tmp}" if $self->{tmp};
  47         my ($tmp, $fh);
  48         do {
  49                 $tmp = _fn_in($self, 'tmp');
  50                 $! = undef;
  51         } while (!sysopen($fh, $tmp, O_CREAT|O_EXCL|O_RDWR) && $!{EEXIST});
  52         print $fh $$strref or die "write failed: $!";
  53         $fh->flush or die "flush failed: $!";
  54         $fh->autoflush(1);
  55         $self->{fh} = $fh;
  56         $self->{tmp} = $tmp;
  57 }
  58
  59 sub abort {
  60         my ($self) = @_;
  61         delete $self->{fh};
  62         my $tmp = delete $self->{tmp} or return;
  63
  64         unlink($tmp) or warn "Failed to unlink $tmp: $!";
  65         undef;
  66 }
  67
  68 sub fh {
  69         my ($self) = @_;
  70         my $fh = $self->{fh} or die "{fh} not open!\n";
  71         seek($fh, 0, SEEK_SET) or die "seek(fh) failed: $!";
  72         sysseek($fh, 0, SEEK_SET) or die "sysseek(fh) failed: $!";
  73         $fh;
  74 }
  75
  76 sub commit {
  77         my ($self) = @_;
  78         $$ == $self->{pid} or return; # no-op in forked child
  79
  80         delete $self->{fh};
  81         my $tmp = delete $self->{tmp} or return;
  82         my $new;
  83         do {
  84                 $new = _fn_in($self, 'new');
  85         } while (!link($tmp, $new) && $!{EEXIST});
  86         my @sn = stat($new) or die "stat $new failed: $!";
  87         my @st = stat($tmp) or die "stat $tmp failed: $!";
  88         if ($st[0] == $sn[0] && $st[1] == $sn[1]) {
  89                 unlink($tmp) or warn "Failed to unlink $tmp: $!";
  90         } else {
  91                 warn "stat($new) and stat($tmp) differ";
  92         }
  93 }
  94
  95 sub DESTROY { commit($_[0]) }
  96
  97 1;