5133778705

Build Type

push

github

Committed by darkdarin

Commit Message

fix: set next watch time at timeout_at for correct timeout promises

Run Details

669 of 1021 relevant lines covered (65.52%)

2.33 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/src/Core/GarbageCollector.php

<?php

namespace Tochka\Promises\Core;

use Carbon\Carbon;
use Illuminate\Support\Collection;
use Illuminate\Support\Facades\DB;
use Tochka\Promises\Core\Support\DaemonWorker;

class GarbageCollector
{
    use DaemonWorker;

    private int $deleteOlderThen;
    /** @var array<string> */
    private array $states;

    private string $promisesTable;
    private string $promiseJobsTable;
    private string $promiseEventsTable;
    private int $promiseChunkSize;
    private int $jobsChunkSize;

    public function __construct(
        int $sleepTime,
        int $deleteOlderThen,
        array $states,
        string $promisesTable,
        string $promiseJobsTable,
        string $promiseEventsTable,
        int $promiseChunkSize = 100,
        int $jobsChunkSize = 500,
    ) {
        $this->sleepTime = $sleepTime;
        $this->deleteOlderThen = $deleteOlderThen;
        $this->states = $states;

        $this->promisesTable = $promisesTable;
        $this->promiseJobsTable = $promiseJobsTable;
        $this->promiseEventsTable = $promiseEventsTable;

        $this->promiseChunkSize = $promiseChunkSize;
        $this->jobsChunkSize = $jobsChunkSize;

        $this->lastIteration = Carbon::minValue();
    }

    public function handle(): void
    {
        $this->daemon(function () {
            $this->clean();
        });
    }

    public function clean(): void
    {
        while (true) {
            $promises = DB::table($this->promisesTable)
                ->select([$this->promiseColumn('id')])
                ->leftJoin(
                    $this->promiseJobsTable,
                    $this->promiseJobsColumn('id'),
                    '=',
                    $this->promiseColumn('parent_job_id')
                )
                ->whereIn($this->promiseColumn('state'), $this->states)
                ->where($this->promiseColumn('updated_at'), '<', Carbon::now()->subSeconds($this->deleteOlderThen))
                ->whereNull($this->promiseJobsColumn('id'))
                ->limit($this->promiseChunkSize)
                ->pluck('id')
                ->all();

            if (empty($promises)) {
                return;
            }

            $this->handlePromiseChunks($promises);

            $this->sleep(0.05);
        }
    }

    private function promiseColumn(string $columnName): string
    {
        return $this->promisesTable . '.' . $columnName;
    }

    private function promiseJobsColumn(string $columnName): string
    {
        return $this->promiseJobsTable . '.' . $columnName;
    }

    /**
     * @param array<int, int> $promiseIds
     * @return bool
     */
    private function handlePromiseChunks(array $promiseIds): bool
    {
        if ($this->paused() || $this->shouldQuit()) {
            return false;
        }

        DB::table($this->promiseJobsTable)
            ->select(['id'])
            ->whereIn('promise_id', $promiseIds)
            ->chunkById(
                $this->jobsChunkSize,
                $this->handleJobsChunks(...)
            );

        DB::table($this->promisesTable)->whereIn('id', $promiseIds)->delete();

        return true;
    }

    /**
     * @param Collection<int, object{id: string}> $jobs
     * @return bool
     */
    private function handleJobsChunks(Collection $jobs): bool
    {
        if ($this->paused() || $this->shouldQuit()) {
            return false;
        }

        $jobsIds = $jobs->pluck('id')->all();

        DB::table($this->promiseEventsTable)->whereIn('job_id', $jobsIds)->delete();
        DB::table($this->promiseJobsTable)->whereIn('id', $jobsIds)->delete();

        return true;
    }
}

1	<?php
2
3	namespace Tochka\Promises\Core;
4
5	use Carbon\Carbon;
6	use Illuminate\Support\Collection;
7	use Illuminate\Support\Facades\DB;
8	use Tochka\Promises\Core\Support\DaemonWorker;
9
10	class GarbageCollector
11	{
12	use DaemonWorker;
13
14	private int $deleteOlderThen;
15	/** @var array<string> */
16	private array $states;
17
18	private string $promisesTable;
19	private string $promiseJobsTable;
20	private string $promiseEventsTable;
21	private int $promiseChunkSize;
22	private int $jobsChunkSize;
23
24	public function __construct(
25	int $sleepTime,
26	int $deleteOlderThen,
27	array $states,
28	string $promisesTable,
29	string $promiseJobsTable,
30	string $promiseEventsTable,
31	int $promiseChunkSize = 100,
32	int $jobsChunkSize = 500,
33	) {
34	$this->sleepTime = $sleepTime;	×
35	$this->deleteOlderThen = $deleteOlderThen;	×
36	$this->states = $states;	×
37
38	$this->promisesTable = $promisesTable;	×
39	$this->promiseJobsTable = $promiseJobsTable;	×
40	$this->promiseEventsTable = $promiseEventsTable;	×
41
42	$this->promiseChunkSize = $promiseChunkSize;	×
43	$this->jobsChunkSize = $jobsChunkSize;	×
44
45	$this->lastIteration = Carbon::minValue();	×
46	}
47
48	public function handle(): void
49	{
50	$this->daemon(function () {	×
51	$this->clean();	×
52	});	×
53	}
54
55	public function clean(): void
56	{
57	while (true) {	×
58	$promises = DB::table($this->promisesTable)	×
59	->select([$this->promiseColumn('id')])	×
60	->leftJoin(	×
61	$this->promiseJobsTable,	×
62	$this->promiseJobsColumn('id'),	×
63	'=',	×
64	$this->promiseColumn('parent_job_id')	×
65	)	×
66	->whereIn($this->promiseColumn('state'), $this->states)	×
67	->where($this->promiseColumn('updated_at'), '<', Carbon::now()->subSeconds($this->deleteOlderThen))	×
68	->whereNull($this->promiseJobsColumn('id'))	×
69	->limit($this->promiseChunkSize)	×
70	->pluck('id')	×
71	->all();	×
72
73	if (empty($promises)) {	×
74	return;	×
75	}
76
77	$this->handlePromiseChunks($promises);	×
78
79	$this->sleep(0.05);	×
80	}
81	}
82
83	private function promiseColumn(string $columnName): string
84	{
85	return $this->promisesTable . '.' . $columnName;	×
86	}
87
88	private function promiseJobsColumn(string $columnName): string
89	{
90	return $this->promiseJobsTable . '.' . $columnName;	×
91	}
92
93	/**
94	* @param array<int, int> $promiseIds
95	* @return bool
96	*/
97	private function handlePromiseChunks(array $promiseIds): bool
98	{
99	if ($this->paused() \|\| $this->shouldQuit()) {	×
100	return false;	×
101	}
102
103	DB::table($this->promiseJobsTable)	×
104	->select(['id'])	×
105	->whereIn('promise_id', $promiseIds)	×
106	->chunkById(	×
107	$this->jobsChunkSize,	×
108	$this->handleJobsChunks(...)	×
109	);	×
110
111	DB::table($this->promisesTable)->whereIn('id', $promiseIds)->delete();	×
112
113	return true;	×
114	}
115
116	/**
117	* @param Collection<int, object{id: string}> $jobs
118	* @return bool
119	*/
120	private function handleJobsChunks(Collection $jobs): bool
121	{
122	if ($this->paused() \|\| $this->shouldQuit()) {	×
123	return false;	×
124	}
125
126	$jobsIds = $jobs->pluck('id')->all();	×
127
128	DB::table($this->promiseEventsTable)->whereIn('job_id', $jobsIds)->delete();	×
129	DB::table($this->promiseJobsTable)->whereIn('id', $jobsIds)->delete();	×
130
131	return true;	×
132	}
133	}

tochka-developers / queue-promises / 5133778705

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous