view release on metacpan or  search on metacpan

lib/Convert/Pheno/HTTP/Jobs.pm  view on Meta::CPAN

    if (-d $final) {
        $status = $publication->{completed};
        _write(path($dir,'status.json'), $status);
        my $marker = path($final,'.convert-pheno-owner');
        unlink $marker or die "Cannot clear publication marker\n" if -e $marker;
    }
    if (-d $staging) {
        remove_tree("$staging", {error => \my $errors});
        die "Cannot clean incomplete export\n" if @$errors;
    }
    unlink $journal or die "Cannot clear publication record\n";
    return $status;
}

sub new {
    my ($class, %args) = @_;
    die "Job storage and worker executable are required\n" unless $args{root} && $args{worker};
    make_path($args{root}, { mode => 0700 });
    my $self = bless { %args, root => _resolved_path($args{root}), queue => [], grants => {}, active => {} }, $class;
    # Recovery may rewrite run state, so only one supervisor may own this store.
    # Keep the handle open for its lifetime; do not unlink the lock file.
    open my $lock, '>>', path($self->{root}, '.supervisor.lock') or die "Cannot lock job storage\n";
    flock($lock, LOCK_EX | LOCK_NB) or die "This job storage is already in use by another Convert-Pheno service\n";
    $self->{storage_lock} = $lock;
    # Desktop supplies its platform-detected CPU limit. Standalone services
    # default to one worker unless the operator explicitly configures a limit.
    $self->{max_allowed_jobs} = _validate_job_limit($ENV{CONVERT_PHENO_JOB_LIMIT} // 1);
    my $settings_file = path($self->{root}, 'scheduler-settings.json');
    my $settings = -f $settings_file ? _read($settings_file) : {maxConcurrentJobs => 1};
    $self->{max_concurrent_jobs} = _validate_job_limit($settings->{maxConcurrentJobs});
    $self->{max_concurrent_jobs} = $self->{max_allowed_jobs}
      if $self->{max_concurrent_jobs} > $self->{max_allowed_jobs};
    my %worker_locks;
    # Preflight all runs before changing any state. A surviving child still owns
    # its run even if its former supervisor's store lock has been released.
    for my $dir (path($self->{root})->children) {
        next unless $dir->basename =~ /\A[a-f0-9]{40}\z/ && -f "$dir/status.json";
        $worker_locks{"$dir"} = _worker_lock($dir);
    }
    for my $dir (path($self->{root})->children) {
        next unless -f "$dir/status.json";
        my $status = eval { _read("$dir/status.json") } or next;
        $status = _recover_publication($dir, $status);
        if ($status->{status} =~ /\A(?:queued|running|cancelling)\z/) {
            $status->{status} = 'interrupted';
            $status->{message} = 'The service stopped before this run completed';
            $status->{finished} = time();
            _write("$dir/status.json", $status);
            remove_tree("$dir/staging");
        }
        unlink "$dir/request.json";
    }
    return $self;
}

sub settings {
    my ($self) = @_;
    return {maxConcurrentJobs => $self->{max_concurrent_jobs}, maxAllowedConcurrentJobs => $self->{max_allowed_jobs}};
}

sub update_settings {
    my ($self, $settings) = @_;
    die "The job service has stopped\n" if $self->{stopped};
    die "Provide only maxConcurrentJobs\n" unless ref($settings) eq 'HASH'
      && keys(%$settings) == 1 && exists $settings->{maxConcurrentJobs};
    my $limit = _validate_job_limit($settings->{maxConcurrentJobs}, $self->{max_allowed_jobs});
    _write(path($self->{root}, 'scheduler-settings.json'), {maxConcurrentJobs => $limit});
    $self->{max_concurrent_jobs} = $limit;
    $self->_next;
    return $self->settings;
}

sub register_file {
    my ($self, $file) = @_;
    die "Select an existing file or directory\n" if !defined($file) || ref($file) || -l $file;
    my $resolved = _resolved_path($file);
    die "Selected location is unavailable\n" unless $resolved && (-f $resolved || -d $resolved);
    my $id = _id();
    $self->{grants}{$id} = $resolved;
    return { id => $id, filename => path($resolved)->basename,
        directory => -d $resolved ? JSON::XS::true : JSON::XS::false,
        bytes => -f $resolved ? -s $resolved : 0 };
}

sub resolve_grant {
    my ($self, $id) = @_;
    die "Unknown or expired file selection; select the file again\n"
      unless defined($id) && !ref($id) && exists $self->{grants}{$id};
    my $file = $self->{grants}{$id};
    die "Selected location changed or is no longer available\n"
      unless !-l $file && -e $file && _same_path(abs_path($file), $file);
    return $file;
}

sub submit {
    my ($self, $request) = @_;
    die "The job service has stopped\n" if $self->{stopped};
    die "Request must be an object\n" unless ref($request) eq 'HASH';
    my %allowed = map { $_ => 1 } qw(conversion input output options destination);
    die "Unknown job field\n" if grep { !$allowed{$_} } keys %$request;
    die "The conversion queue is full\n" if @{$self->{queue}} >= 16;
    my $conversion = $request->{conversion};
    my ($route) = grep { defined($conversion) && !ref($conversion) && $_->{id} eq $conversion } @{catalog()->{data}};
    die "Unknown conversion\n" unless $route;
    die "$route->{unavailableReason}\n" unless $route->{available};
    my $input = $request->{input};
    die "Provide JSON data or selected input files\n" unless ref($input) eq 'HASH'
      && keys(%$input) == 1 && (exists $input->{data} || ref($input->{files}) eq 'HASH');
    my %files;
    for my $role (keys %{$input->{files} || {}}) {
        die "Selected files must be an array\n" unless ref($input->{files}{$role}) eq 'ARRAY';
        $files{$role} = [map {
            my $file = $self->resolve_grant($_);
            +{path => $file, filename => path($file)->basename};
        } @{$input->{files}{$role}}];
    }
    my $destination = exists $request->{destination} ? $self->resolve_grant($request->{destination}) : undef;
    die "Output destination must be a directory\n" if defined($destination) && !-d $destination;
    my $id = _id();
    my $dir = path($self->{root}, $id);
    $dir->mkpath({mode => 0700});