lib/PublicInbox/Emergency.pm

   1 # Copyright (C) 2016-2021 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 v5.10.1;
   8 use Fcntl qw(:DEFAULT SEEK_SET);
   9 use Sys::Hostname qw(hostname);
  10 use IO::Handle; # ->flush
  11 use Errno qw(EEXIST);
  12
  13 sub new {
  14         my ($class, $dir) = @_;
  15
  16         foreach (qw(new tmp cur)) {
  17                 my $d = "$dir/$_";
  18                 next if -d $d;
  19                 require File::Path;
  20                 if (!File::Path::mkpath($d) && !-d $d) {
  21                         die "failed to mkpath($d): $!\n";
  22                 }
  23         }
  24         bless { dir => $dir, t => 0 }, $class;
  25 }
  26
  27 sub _fn_in {
  28         my ($self, $pid, $dir) = @_;
  29         my $host = $self->{short_host} //= (split(/\./, hostname))[0];
  30         my $now = time;
  31         my $n;
  32         if ($self->{t} != $now) {
  33                 $self->{t} = $now;
  34                 $n = $self->{cnt} = 0;
  35         } else {
  36                 $n = ++$self->{cnt};
  37         }
  38         "$self->{dir}/$dir/$self->{t}.$pid"."_$n.$host";
  39 }
  40
  41 sub prepare {
  42         my ($self, $strref) = @_;
  43         my $pid = $$;
  44         my $tmp_key = "tmp.$pid";
  45         die "already in transaction: $self->{$tmp_key}" if $self->{$tmp_key};
  46         my ($tmp, $fh);
  47         do {
  48                 $tmp = _fn_in($self, $pid, 'tmp');
  49                 $! = undef;
  50         } while (!sysopen($fh, $tmp, O_CREAT|O_EXCL|O_RDWR) and $! == EEXIST);
  51         print $fh $$strref or die "write failed: $!";
  52         $fh->flush or die "flush failed: $!";
  53         $self->{fh} = $fh;
  54         $self->{$tmp_key} = $tmp;
  55 }
  56
  57 sub abort {
  58         my ($self) = @_;
  59         delete $self->{fh};
  60         my $tmp = delete $self->{"tmp.$$"} or return;
  61         unlink($tmp) or warn "Failed to unlink $tmp: $!";
  62         undef;
  63 }
  64
  65 sub fh {
  66         my ($self) = @_;
  67         my $fh = $self->{fh} or die "{fh} not open!\n";
  68         seek($fh, 0, SEEK_SET) or die "seek(fh) failed: $!";
  69         sysseek($fh, 0, SEEK_SET) or die "sysseek(fh) failed: $!";
  70         $fh;
  71 }
  72
  73 sub commit {
  74         my ($self) = @_;
  75         my $pid = $$;
  76         my $tmp = delete $self->{"tmp.$pid"} or return;
  77         delete $self->{fh};
  78         my ($new, $ok);
  79         do {
  80                 $new = _fn_in($self, $pid, 'new');
  81         } while (!($ok = link($tmp, $new)) && $! == EEXIST);
  82         die "link($tmp, $new): $!" unless $ok;
  83         unlink($tmp) or warn "Failed to unlink $tmp: $!";
  84 }
  85
  86 sub DESTROY { commit($_[0]) }
  87
  88 1;