media-cloud-sync / includes / sdk / s3 / Aws / RetryMiddlewareV2.php

RetryMiddlewareV2.php in Media Cloud Sync 1.2.12, at includes/sdk/s3/Aws/RetryMiddlewareV2.php

236 lines 10.5 KB

Raw Download Zip

1	<?php
2
3	namespace Dudlewebs\WPMCS\s3\Aws;
4
5	use Dudlewebs\WPMCS\s3\Aws\Exception\AwsException;
6	use Dudlewebs\WPMCS\s3\Aws\Retry\ConfigurationInterface;
7	use Dudlewebs\WPMCS\s3\Aws\Retry\QuotaManager;
8	use Dudlewebs\WPMCS\s3\Aws\Retry\RateLimiter;
9	use Dudlewebs\WPMCS\s3\Aws\Retry\RetryHelperTrait;
10	use Dudlewebs\WPMCS\s3\GuzzleHttp\Exception\RequestException;
11	use Dudlewebs\WPMCS\s3\GuzzleHttp\Promise;
12	use Dudlewebs\WPMCS\s3\Psr\Http\Message\RequestInterface;
13	/**
14	* Middleware that retries failures. V2 implementation that supports 'standard'
15	* and 'adaptive' modes.
16	*
17	* @internal
18	*/
19	class RetryMiddlewareV2
20	{
21	use RetryHelperTrait;
22	private static $standardThrottlingErrors = ['Throttling' => \true, 'ThrottlingException' => \true, 'ThrottledException' => \true, 'RequestThrottledException' => \true, 'TooManyRequestsException' => \true, 'ProvisionedThroughputExceededException' => \true, 'TransactionInProgressException' => \true, 'RequestLimitExceeded' => \true, 'BandwidthLimitExceeded' => \true, 'LimitExceededException' => \true, 'RequestThrottled' => \true, 'SlowDown' => \true, 'PriorRequestNotComplete' => \true, 'EC2ThrottledException' => \true];
23	private static $standardTransientErrors = ['RequestTimeout' => \true, 'RequestTimeoutException' => \true];
24	private static $standardTransientStatusCodes = [500 => \true, 502 => \true, 503 => \true, 504 => \true];
25	private $collectStats;
26	private $decider;
27	private $delayer;
28	private $maxAttempts;
29	private $maxBackoff;
30	private $mode;
31	private $nextHandler;
32	private $options;
33	private $quotaManager;
34	private $rateLimiter;
35	public static function wrap($config, $options)
36	{
37	return function (callable $handler) use($config, $options) {
38	return new static($config, $handler, $options);
39	};
40	}
41	public static function createDefaultDecider(QuotaManager $quotaManager, $maxAttempts = 3, $options = [])
42	{
43	$retryCurlErrors = [];
44	if (\extension_loaded('curl')) {
45	$retryCurlErrors[\CURLE_RECV_ERROR] = \true;
46	}
47	return function ($attempts, CommandInterface $command, $result) use($options, $quotaManager, $retryCurlErrors, $maxAttempts) {
48	// Release retry tokens back to quota on a successful result
49	$quotaManager->releaseToQuota($result);
50	// Allow command-level option to override this value
51	// # of attempts = # of retries + 1
52	$maxAttempts = null !== $command['@retries'] ? $command['@retries'] + 1 : $maxAttempts;
53	$isRetryable = self::isRetryable($result, $retryCurlErrors, $options);
54	if ($isRetryable) {
55	// Retrieve retry tokens and check if quota has been exceeded
56	if (!$quotaManager->hasRetryQuota($result)) {
57	return \false;
58	}
59	if ($attempts >= $maxAttempts) {
60	if (!empty($result) && $result instanceof AwsException) {
61	$result->setMaxRetriesExceeded();
62	}
63	return \false;
64	}
65	}
66	return $isRetryable;
67	};
68	}
69	public function __construct(ConfigurationInterface $config, callable $handler, $options = [])
70	{
71	$this->options = $options;
72	$this->maxAttempts = $config->getMaxAttempts();
73	$this->mode = $config->getMode();
74	$this->nextHandler = $handler;
75	$this->quotaManager = new QuotaManager();
76	$this->maxBackoff = isset($options['max_backoff']) ? $options['max_backoff'] : 20000;
77	$this->collectStats = isset($options['collect_stats']) ? (bool) $options['collect_stats'] : \false;
78	$this->decider = isset($options['decider']) ? $options['decider'] : self::createDefaultDecider($this->quotaManager, $this->maxAttempts, $options);
79	$this->delayer = isset($options['delayer']) ? $options['delayer'] : function ($attempts) {
80	return $this->exponentialDelayWithJitter($attempts);
81	};
82	if ($this->mode === 'adaptive') {
83	$this->rateLimiter = isset($options['rate_limiter']) ? $options['rate_limiter'] : new RateLimiter();
84	}
85	}
86	public function __invoke(CommandInterface $cmd, RequestInterface $req)
87	{
88	$decider = $this->decider;
89	$delayer = $this->delayer;
90	$handler = $this->nextHandler;
91	$attempts = 1;
92	$monitoringEvents = [];
93	$requestStats = [];
94	$req = $this->addRetryHeader($req, 0, 0);
95	$callback = function ($value) use($handler, $cmd, $req, $decider, $delayer, &$attempts, &$requestStats, &$monitoringEvents, &$callback) {
96	if ($this->mode === 'adaptive') {
97	$this->rateLimiter->updateSendingRate($this->isThrottlingError($value));
98	}
99	$this->updateHttpStats($value, $requestStats);
100	if ($value instanceof MonitoringEventsInterface) {
101	$reversedEvents = \array_reverse($monitoringEvents);
102	$monitoringEvents = \array_merge($monitoringEvents, $value->getMonitoringEvents());
103	foreach ($reversedEvents as $event) {
104	$value->prependMonitoringEvent($event);
105	}
106	}
107	if ($value instanceof \Exception \|\| $value instanceof \Throwable) {
108	if (!$decider($attempts, $cmd, $value)) {
109	return Promise\Create::rejectionFor($this->bindStatsToReturn($value, $requestStats));
110	}
111	} elseif ($value instanceof ResultInterface && !$decider($attempts, $cmd, $value)) {
112	return $this->bindStatsToReturn($value, $requestStats);
113	}
114	$delayBy = $delayer($attempts++);
115	$cmd['@http']['delay'] = $delayBy;
116	if ($this->collectStats) {
117	$this->updateStats($attempts - 1, $delayBy, $requestStats);
118	}
119	// Update retry header with retry count and delayBy
120	$req = $this->addRetryHeader($req, $attempts - 1, $delayBy);
121	// Get token from rate limiter, which will sleep if necessary
122	if ($this->mode === 'adaptive') {
123	$this->rateLimiter->getSendToken();
124	}
125	return $handler($cmd, $req)->then($callback, $callback);
126	};
127	// Get token from rate limiter, which will sleep if necessary
128	if ($this->mode === 'adaptive') {
129	$this->rateLimiter->getSendToken();
130	}
131	return $handler($cmd, $req)->then($callback, $callback);
132	}
133	/**
134	* Amount of milliseconds to delay as a function of attempt number
135	*
136	* @param $attempts
137	* @return mixed
138	*/
139	public function exponentialDelayWithJitter($attempts)
140	{
141	$rand = \mt_rand() / \mt_getrandmax();
142	return \min(1000 * $rand * \pow(2, $attempts), $this->maxBackoff);
143	}
144	private static function isRetryable($result, $retryCurlErrors, $options = [])
145	{
146	$errorCodes = self::$standardThrottlingErrors + self::$standardTransientErrors;
147	if (!empty($options['transient_error_codes']) && \is_array($options['transient_error_codes'])) {
148	foreach ($options['transient_error_codes'] as $code) {
149	$errorCodes[$code] = \true;
150	}
151	}
152	if (!empty($options['throttling_error_codes']) && \is_array($options['throttling_error_codes'])) {
153	foreach ($options['throttling_error_codes'] as $code) {
154	$errorCodes[$code] = \true;
155	}
156	}
157	$statusCodes = self::$standardTransientStatusCodes;
158	if (!empty($options['status_codes']) && \is_array($options['status_codes'])) {
159	foreach ($options['status_codes'] as $code) {
160	$statusCodes[$code] = \true;
161	}
162	}
163	if (!empty($options['curl_errors']) && \is_array($options['curl_errors'])) {
164	foreach ($options['curl_errors'] as $code) {
165	$retryCurlErrors[$code] = \true;
166	}
167	}
168	if ($result instanceof \Exception \|\| $result instanceof \Throwable) {
169	$isError = \true;
170	} else {
171	$isError = \false;
172	}
173	if (!$isError) {
174	if (!isset($result['@metadata']['statusCode'])) {
175	return \false;
176	}
177	return isset($statusCodes[$result['@metadata']['statusCode']]);
178	}
179	if (!$result instanceof AwsException) {
180	return \false;
181	}
182	if ($result->isConnectionError()) {
183	return \true;
184	}
185	if (!empty($errorCodes[$result->getAwsErrorCode()])) {
186	return \true;
187	}
188	if (!empty($statusCodes[$result->getStatusCode()])) {
189	return \true;
190	}
191	if (\count($retryCurlErrors) && ($previous = $result->getPrevious()) && $previous instanceof RequestException) {
192	if (\method_exists($previous, 'getHandlerContext')) {
193	$context = $previous->getHandlerContext();
194	return !empty($context['errno']) && isset($retryCurlErrors[$context['errno']]);
195	}
196	$message = $previous->getMessage();
197	foreach (\array_keys($retryCurlErrors) as $curlError) {
198	if (\strpos($message, 'cURL error ' . $curlError . ':') === 0) {
199	return \true;
200	}
201	}
202	}
203	// Check error shape for the retryable trait
204	if (!empty($errorShape = $result->getAwsErrorShape())) {
205	$definition = $errorShape->toArray();
206	if (!empty($definition['retryable'])) {
207	return \true;
208	}
209	}
210	return \false;
211	}
212	private function isThrottlingError($result)
213	{
214	if ($result instanceof AwsException) {
215	// Check pre-defined throttling errors
216	$throttlingErrors = self::$standardThrottlingErrors;
217	if (!empty($this->options['throttling_error_codes']) && \is_array($this->options['throttling_error_codes'])) {
218	foreach ($this->options['throttling_error_codes'] as $code) {
219	$throttlingErrors[$code] = \true;
220	}
221	}
222	if (!empty($result->getAwsErrorCode()) && !empty($throttlingErrors[$result->getAwsErrorCode()])) {
223	return \true;
224	}
225	// Check error shape for the throttling trait
226	if (!empty($errorShape = $result->getAwsErrorShape())) {
227	$definition = $errorShape->toArray();
228	if (!empty($definition['retryable']['throttling'])) {
229	return \true;
230	}
231	}
232	}
233	return \false;
234	}
235	}
236