ipc: more error-checking for sendmsg(..., MSG_EOR)

Introduce `sendmsg_eor' for error checking to ensure we don't hit
short writes when writing to SOCK_SEQPACKET sockets.  Reducing
imports of MSG_EOR will also make it easier for us to drop it in
the future if certain OSes end up being too buggy with it (as was
the case with OpenBSD 7.3).  It'll also make it easier to switch
to SOCK_DGRAM in the future if necessary, since Unix datagram
sockets seem reliable in practice.

diff --git a/lib/PublicInbox/IPC.pm b/lib/PublicInbox/IPC.pm
index efbb210469366444cac48a7a597589cf2ebb7f32..65e081638bf766bfe1464bf3745af2b5da6d6d49 100644 (file)
--- a/lib/PublicInbox/IPC.pm
+++ b/lib/PublicInbox/IPC.pm
@@ -398,16 +398,22 @@ sub wq_broadcast {
        croak "@exc" if @exc;
 }
 
+sub sendmsg_eor ($$$;$) {
+       my $n = $send_cmd->($_[0], $_[1], $_[2], MSG_EOR, $_[3] // 50) //
+               return;
+       $n == length($_[2]) ? $n : croak('sendmsg('.length($_[2])." > $n)");
+}
+
 sub stream_in_full ($$$) {
        my ($s1, $io, $buf) = @_;
        socketpair(my $r, my $w, AF_UNIX, SOCK_STREAM, 0);
-       my $n = $send_cmd->($s1, [ $r ],
-                       ipc_freeze(['do_sock_stream', length($buf)]),
-                       MSG_EOR) // croak "sendmsg: $!";
+       my $n = sendmsg_eor($s1, [ $r ],
+                       ipc_freeze(['do_sock_stream', length($buf)]))
+               // croak "sendmsg: $!";
        undef $r;
        $n = $send_cmd->($w, $io, $buf, 0) // croak "sendmsg: $!";
        print $w substr($buf, $n) if $n < length($buf); # need > 2G on Linux
-       close $w; # autodies
+       close $w; # autodies if print failed
 }
 
 sub wq_io_do { # always async
@@ -416,9 +422,9 @@ sub wq_io_do { # always async
        my $buf = ipc_freeze([$sub, @args]);
        if (length($buf) > $MY_MAX_ARG_LEN) {
                stream_in_full($s1, $io, $buf);
+       } elsif (defined sendmsg_eor($s1, $io, $buf)) {
+               # success
        } else {
-               my $n = $send_cmd->($s1, $io, $buf, MSG_EOR);
-               return if defined($n); # likely
                $!{ETOOMANYREFS} and croak "sendmsg: $! (check RLIMIT_NOFILE)";
                $!{EMSGSIZE} ? stream_in_full($s1, $io, $buf) :
                        croak("sendmsg: $!");
@@ -458,13 +464,13 @@ sub wq_nonblock_do { # always async
        my $buf = ipc_freeze([$sub, @args]);
        if ($self->{wqb}) { # saturated once, assume saturated forever
                $self->{wqb}->flush_send($buf);
-       } elsif (!defined $send_cmd->($self->{-wq_s1}, [], $buf, MSG_EOR)) {
-               if ($!{EAGAIN} || $!{ENOBUFS} || $!{ENOMEM}) {
-                       PublicInbox::WQBlocked->new($self, $buf);
-               } else {
-                       croak "sendmsg: $!";
-               }
-       } # else success
+       } elsif (defined sendmsg_eor($self->{-wq_s1}, [], $buf)) {
+               # success!
+       } elsif ($!{EAGAIN} || $!{ENOBUFS} || $!{ENOMEM}) {
+               PublicInbox::WQBlocked->new($self, $buf);
+       } else {
+               croak "sendmsg: $!";
+       }
 }
 
 sub _wq_worker_start {

diff --git a/lib/PublicInbox/LEI.pm b/lib/PublicInbox/LEI.pm
index cc62c6cfbfb8e638ca221c11b01ab1290bd77966..9cf77fe216808b6b316c4071efabd324e83d927d 100644 (file)
--- a/lib/PublicInbox/LEI.pm
+++ b/lib/PublicInbox/LEI.pm
@@ -11,7 +11,7 @@ use parent qw(PublicInbox::DS PublicInbox::LeiExternal
        PublicInbox::LeiQuery);
 use autodie qw(bind chdir listen open pipe socket socketpair syswrite unlink);
 use Getopt::Long ();
-use Socket qw(AF_UNIX SOCK_SEQPACKET pack_sockaddr_un MSG_EOR);
+use Socket qw(AF_UNIX SOCK_SEQPACKET pack_sockaddr_un);
 use Errno qw(EPIPE EAGAIN ECONNREFUSED ENOENT ECONNRESET EINTR);
 use Cwd qw(getcwd);
 use POSIX qw(strftime);
@@ -1057,9 +1057,9 @@ sub start_mua {
 
 sub send_exec_cmd { # tell script/lei to execute a command
        my ($self, $io, $cmd, $env) = @_;
-       $PublicInbox::IPC::send_cmd->(
+       PublicInbox::IPC::sendmsg_eor(
                        $self->{sock} // die('lei client gone'),
-                       $io, exec_buf($cmd, $env), MSG_EOR) //
+                       $io, exec_buf($cmd, $env)) //
                Carp::croak("sendmsg: $!");
 }
 

diff --git a/lib/PublicInbox/WQBlocked.pm b/lib/PublicInbox/WQBlocked.pm
index e733f446f21f87f18ee9e83265c564cc4b802c54..d01d753cf6d04d980e0efbf014cf9cf77f9f8c6e 100644 (file)
--- a/lib/PublicInbox/WQBlocked.pm
+++ b/lib/PublicInbox/WQBlocked.pm
@@ -6,7 +6,6 @@ package PublicInbox::WQBlocked;
 use v5.12;
 use parent qw(PublicInbox::DS);
 use PublicInbox::Syscall qw(EPOLLOUT EPOLLONESHOT);
-use Socket qw(MSG_EOR);
 use PublicInbox::IPC;
 use Carp ();
 
@@ -22,13 +21,12 @@ sub flush_send {
        while (defined(my $buf = shift @{$self->{msgq}})) {
                if (ref($buf) eq 'CODE') {
                        $buf->($self); # could be \&PublicInbox::DS::close
+               } elsif (defined(PublicInbox::IPC::sendmsg_eor(
+                                       $self->{sock}, [], $buf))) {
+                       # success
                } else {
-                       my $wq_s1 = $self->{sock};
-                       my $n = $PublicInbox::IPC::send_cmd->($wq_s1, [], $buf,
-                                                               MSG_EOR);
-                       next if defined($n);
                        if ($!{EAGAIN}) {
-                               PublicInbox::DS::epwait($wq_s1,
+                               PublicInbox::DS::epwait($self->{sock},
                                                        EPOLLOUT|EPOLLONESHOT);
                        } elsif ($!{ENOBUFS} || $!{ENOMEM}) {
                                PublicInbox::DS::add_uniq_timer($self + 0,

diff --git a/lib/PublicInbox/XapClient.pm b/lib/PublicInbox/XapClient.pm
index 88638aed9e575e654c71b835191d0919a917bee7..7f410ff97acc1ab34007f850e5ef5a124ff40fb1 100644 (file)
--- a/lib/PublicInbox/XapClient.pm
+++ b/lib/PublicInbox/XapClient.pm
@@ -9,7 +9,8 @@
 package PublicInbox::XapClient;
 use v5.12;
 use PublicInbox::Spawn qw(spawn);
-use Socket qw(AF_UNIX SOCK_SEQPACKET MSG_EOR);
+use Carp qw(croak);
+use Socket qw(AF_UNIX SOCK_SEQPACKET);
 use PublicInbox::IPC;
 use autodie qw(pipe socketpair);
 our $tries = -1; # set to zero by read-only daemon
@@ -17,10 +18,8 @@ our $tries = -1; # set to zero by read-only daemon
 sub mkreq {
        my ($self, $io, @arg) = @_;
        my $buf = join("\0", @arg, '');
-       my $n = $PublicInbox::IPC::send_cmd->($self->{io},
-                               $io, $buf, MSG_EOR, $tries)
-                               // die "send_cmd: $!";
-       $n == length($buf) or die "send_cmd: $n != ".length($buf);
+       PublicInbox::IPC::sendmsg_eor($self->{io}, $io, $buf, $tries) //
+               croak "sendmsg_eor: $!";
 }
 
 sub start_helper (@) {

diff --git a/t/lei-daemon.t b/t/lei-daemon.t
index 0097c6f0a68c9a6535e2e260120292ae0d8b41f6..f1b052ec2c8db0f43f2f63d3f4d27037956e0f31 100644 (file)
--- a/t/lei-daemon.t
+++ b/t/lei-daemon.t
@@ -2,15 +2,10 @@
 # Copyright (C) all contributors <meta@public-inbox.org>
 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
 use strict; use v5.10.1; use PublicInbox::TestCommon;
-use Socket qw(AF_UNIX SOCK_SEQPACKET pack_sockaddr_un MSG_EOR);
+use Socket qw(AF_UNIX SOCK_SEQPACKET pack_sockaddr_un);
+require PublicInbox::IPC;
 
 test_lei({ daemon_only => 1 }, sub {
-       my $send_cmd = PublicInbox::Spawn->can('send_cmd4') // do {
-               require PublicInbox::Syscall;
-               PublicInbox::Syscall->can('send_cmd4');
-       };
-       $send_cmd or BAIL_OUT 'started testing lei-daemon w/o send_cmd4!';
-
        my $sock = "$ENV{XDG_RUNTIME_DIR}/lei/5.seq.sock";
        my $err_log = "$ENV{XDG_RUNTIME_DIR}/lei/errors.log";
        lei_ok('daemon-pid');
@@ -36,7 +31,8 @@ test_lei({ daemon_only => 1 }, sub {
                        socket(my $c, AF_UNIX, SOCK_SEQPACKET, 0) or
                                                        BAIL_OUT "socket: $!";
                        connect($c, $addr) or BAIL_OUT "connect: $!";
-                       $send_cmd->($c, [ $null, $null, $null ], 'hi', MSG_EOR);
+                       PublicInbox::IPC::sendmsg_eor($c,
+                                               [ $null, $null, $null ], 'hi');
                }
                lei_ok('daemon-pid');
                chomp($pid = $lei_out);

diff --git a/t/xap_helper.t b/t/xap_helper.t
index 4e7b7f017071b544391e41c01406b52691ca6f81..c5f1411f42bc1dad6f32ce36afd3007f198b41e5 100644 (file)
--- a/t/xap_helper.t
+++ b/t/xap_helper.t
@@ -5,7 +5,7 @@ use v5.12;
 use PublicInbox::TestCommon;
 require_mods(qw(DBD::SQLite Xapian +SCM_RIGHTS)); # TODO: FIFO support?
 use PublicInbox::Spawn qw(spawn);
-use Socket qw(AF_UNIX SOCK_SEQPACKET SOCK_STREAM MSG_EOR);
+use Socket qw(AF_UNIX SOCK_SEQPACKET SOCK_STREAM);
 require PublicInbox::AutoReap;
 use PublicInbox::IPC;
 require PublicInbox::XapClient;
@@ -91,8 +91,8 @@ my $doreq = sub {
        pipe(my $x, my $y);
        my $buf = join("\0", @arg, '');
        my @io = ($y, $err);
-       my $n = $PublicInbox::IPC::send_cmd->($s, \@io, $buf, MSG_EOR) //
-               xbail "send: $!";
+       my $n = PublicInbox::IPC::sendmsg_eor($s, \@io, $buf) //
+               xbail "sendmsg: $!";
        my $exp = length($buf);
        $exp == $n or xbail "req @arg sent short ($n != $exp)";
        $x;