view release on metacpan or  search on metacpan

lib/Developer/Dashboard/Collector.pm  view on Meta::CPAN

    $self->_atomic_write_text( $paths->{last_run}, $timestamp . "\n" );

    my $written = $self->update_status(
        $name,
        sub {
            my ($previous) = @_;
            my %extra = defined $status_callback ? %{ $status_callback->($previous) || {} } : ();
            return {
                %{$previous},
                %extra,
                enabled          => exists $result{enabled} ? $result{enabled} : 1,
                running          => exists $result{running} ? $result{running} : ( exists $extra{running} ? $extra{running} : 0 ),
                active_runs      => exists $result{active_runs} ? $result{active_runs} : ( exists $extra{active_runs} ? $extra{active_runs} : ( $previous->{active_runs} || 0 ) ),
                last_run         => $timestamp,
                last_completed_at=> $timestamp,
                last_exit_code   => $result{exit_code},
                last_success     => $result{exit_code} ? 0 : 1,
                last_success_at  => $result{exit_code} ? ( $previous->{last_success_at} || undef ) : $timestamp,
                last_failure_at  => $result{exit_code} ? $timestamp : ( $previous->{last_failure_at} || undef ),
                last_started_at  => $result{started_at} || $previous->{last_started_at},
                output_format    => $result{output_format},
                timed_out        => $result{timed_out} ? 1 : 0,
            };
        }
    );
    $self->append_log_entry(
        $name,
        happened_at => $timestamp,
        exit_code   => $result{exit_code},
        timed_out   => $result{timed_out},
        stdout      => $result{stdout},
        stderr      => $result{stderr},
    );
    return $written;
}

# write_status($name, $status)
# Merges and persists collector status metadata without rewriting outputs.
# Input: collector name string and partial status hash reference.
# Output: written status file path.
sub write_status {
    my ( $self, $name, $status ) = @_;
    return $self->update_status(
        $name,
        sub {
            my ($existing) = @_;
            return {
                %{$existing},
                %{ $status || {} },
            };
        }
    );
}

# update_status($name, $callback)
# Updates one collector status file under an exclusive lock so concurrent
# workers can safely merge active-run counters and lifecycle fields.
# Input: collector name string and callback that receives the current status
# hash reference and returns a replacement hash reference.
# Output: written status file path.
sub update_status {
    my ( $self, $name, $callback ) = @_;
    die 'Missing collector name' if !defined $name || $name eq '';
    die 'Missing collector status callback' if ref($callback) ne 'CODE';
    my $paths = $self->collector_paths($name);
    my $lock = File::Spec->catfile( $paths->{dir}, '.status.lock' );
    open my $lock_fh, '>>', $lock or die "Unable to open $lock: $!";
    flock( $lock_fh, LOCK_EX ) or die "Unable to lock $lock: $!";
    my $existing = $self->_read_status_file( $paths->{status} ) || {};
    my $next = $callback->($existing);
    die 'Collector status callback must return a hash reference' if ref($next) ne 'HASH';
    my %merged = (
        %{$next},
        name             => $name,
        updated_at_epoch => time,
    );
    return $self->_atomic_write_json( $paths->{status}, \%merged );
}

# mark_run_started($name, $status)
# Records one collector execution start while keeping an accurate active-run
# count for singleton and bounded-multiple loop modes.
# Input: collector name string and optional partial status hash reference.
# Output: written status file path.
sub mark_run_started {
    my ( $self, $name, $status ) = @_;
    return $self->update_status(
        $name,
        sub {
            my ($existing) = @_;
            my $active_runs = ( $existing->{active_runs} || 0 ) + 1;
            return {
                %{$existing},
                %{ $status || {} },
                running         => 1,
                active_runs     => $active_runs,
                last_started_at => ( $status || {} )->{last_started_at} || $existing->{last_started_at},
            };
        }
    );
}

# mark_run_finished($name, %result)
# Records one collector execution completion while decrementing the active-run
# count under lock so overlapping runs cannot clear another live worker.
# Input: collector name string plus result fields such as exit_code/stdout/stderr.
# Output: written status file path.
sub mark_run_finished {
    my ( $self, $name, %result ) = @_;
    return $self->write_result(
        $name,
        %result,
        status_callback => sub {
            my ($previous) = @_;
            my $active_runs = $previous->{active_runs} || 0;
            $active_runs-- if $active_runs > 0;
            return {
                active_runs => $active_runs,
                running     => $active_runs > 0 ? 1 : 0,
            };
        },