Merge pull request #2665 from pixelfed/staging

Update federation pipeline, add locks

CHANGELOG.md

@@ -45,6 +45,7 @@
 - Updated InboxPipeline, fail earlier for invalid public keys. Fixes ([#2648](https://github.com/pixelfed/pixelfed/issues/2648)). ([d1c5e9b8](https://github.com/pixelfed/pixelfed/commit/d1c5e9b8))
 - Updated Status model, refactor liked and shared methods to fix cache invalidation bug. ([f05c3b66](https://github.com/pixelfed/pixelfed/commit/f05c3b66))
 - Updated Timeline component, add inline reports modal. ([e64b4bd3](https://github.com/pixelfed/pixelfed/commit/e64b4bd3))
+- Updated federation pipeline, add locks. ([ddc76887](https://github.com/pixelfed/pixelfed/commit/ddc76887))
 -  ([](https://github.com/pixelfed/pixelfed/commit/))
 
 ## [v0.10.10 (2021-01-28)](https://github.com/pixelfed/pixelfed/compare/v0.10.9...v0.10.10)

app/Jobs/InboxPipeline/InboxValidator.php

@@ -53,6 +53,15 @@ class InboxValidator implements ShouldQueue
 
         $profile = Profile::whereNull('domain')->whereUsername($username)->first();
 
+        if(isset($payload['id'])) {
+            $lockKey = hash('sha256', $payload['id']);
+            if(Cache::get($lockKey) !== null) {
+                // Job processed already
+                return 1;
+            }
+            Cache::put($lockKey, 1, 300);
+        }
+
         if(!isset($headers['signature']) || !isset($headers['date'])) {
             return;
         }

app/Jobs/InboxPipeline/InboxWorker.php

@@ -49,6 +49,15 @@ class InboxWorker implements ShouldQueue
         $headers = $this->headers;
         $payload = json_decode($this->payload, true, 8);
 
+        if(isset($payload['id'])) {
+            $lockKey = hash('sha256', $payload['id']);
+            if(Cache::get($lockKey) !== null) {
+                // Job processed already
+                return 1;
+            }
+            Cache::put($lockKey, 1, 300);
+        }
+
         if(!isset($headers['signature']) || !isset($headers['date'])) {
             return;
         }

app/Util/ActivityPub/Helpers.php

@@ -346,27 +346,41 @@ class Helpers {
 			$reply_to = null;
 		}
 		$ts = is_array($res['published']) ? $res['published'][0] : $res['published'];
-		$status = DB::transaction(function() use($profile, $res, $url, $ts, $reply_to, $cw, $scope, $id) {
+
-			$status = new Status;
+		$statusLockKey = 'helpers:status-lock:' . hash('sha256', $res['id']);
-			$status->profile_id = $profile->id;
+		$status = Cache::lock($statusLockKey)
-			$status->url = isset($res['url']) ? $res['url'] : $url;
+			->get(function () use(
-			$status->uri = isset($res['url']) ? $res['url'] : $url;
+				$profile, 
-			$status->object_url = $id;
+				$res, 
-			$status->caption = strip_tags($res['content']);
+				$url, 
-			$status->rendered = Purify::clean($res['content']);
+				$ts, 
-			$status->created_at = Carbon::parse($ts);
+				$reply_to, 
-			$status->in_reply_to_id = $reply_to;
+				$cw, 
-			$status->local = false;
+				$scope, 
-			$status->is_nsfw = $cw;
+				$id
-			$status->scope = $scope;
+		) {
-			$status->visibility = $scope;
+			return DB::transaction(function() use($profile, $res, $url, $ts, $reply_to, $cw, $scope, $id) {
-			$status->cw_summary = $cw == true && isset($res['summary']) ?
+				$status = new Status;
-				Purify::clean(strip_tags($res['summary'])) : null;
+				$status->profile_id = $profile->id;
-			$status->save();
+				$status->url = isset($res['url']) ? $res['url'] : $url;
-			if($reply_to == null) {
+				$status->uri = isset($res['url']) ? $res['url'] : $url;
-				self::importNoteAttachment($res, $status);
+				$status->object_url = $id;
-			}
+				$status->caption = strip_tags($res['content']);
-			return $status;
+				$status->rendered = Purify::clean($res['content']);
+				$status->created_at = Carbon::parse($ts);
+				$status->in_reply_to_id = $reply_to;
+				$status->local = false;
+				$status->is_nsfw = $cw;
+				$status->scope = $scope;
+				$status->visibility = $scope;
+				$status->cw_summary = $cw == true && isset($res['summary']) ?
+					Purify::clean(strip_tags($res['summary'])) : null;
+				$status->save();
+				if($reply_to == null) {
+					self::importNoteAttachment($res, $status);
+				}
+				return $status;
+			});
 		});
 
 		return $status;
@@ -458,25 +472,28 @@ class Helpers {
 
 			$profile = Profile::whereRemoteUrl($res['id'])->first();
 			if(!$profile) {
-				$profile = DB::transaction(function() use($domain, $webfinger, $res, $runJobs) {
+				$profileLockKey = 'helpers:profile-lock:' . hash('sha256', $res['id']);
-					$profile = new Profile();
+				$profile = Cache::lock($profileLockKey)->get(function () use($domain, $webfinger, $res, $runJobs) {
-					$profile->domain = strtolower($domain);
+					return DB::transaction(function() use($domain, $webfinger, $res, $runJobs) {
-					$profile->username = strtolower(Purify::clean($webfinger));
+						$profile = new Profile();
-					$profile->name = isset($res['name']) ? Purify::clean($res['name']) : 'user';
+						$profile->domain = strtolower($domain);
-					$profile->bio = isset($res['summary']) ? Purify::clean($res['summary']) : null;
+						$profile->username = strtolower(Purify::clean($webfinger));
-					$profile->sharedInbox = isset($res['endpoints']) && isset($res['endpoints']['sharedInbox']) ? $res['endpoints']['sharedInbox'] : null;
+						$profile->name = isset($res['name']) ? Purify::clean($res['name']) : 'user';
-					$profile->inbox_url = strtolower($res['inbox']);
+						$profile->bio = isset($res['summary']) ? Purify::clean($res['summary']) : null;
-					$profile->outbox_url = strtolower($res['outbox']);
+						$profile->sharedInbox = isset($res['endpoints']) && isset($res['endpoints']['sharedInbox']) ? $res['endpoints']['sharedInbox'] : null;
-					$profile->remote_url = strtolower($res['id']);
+						$profile->inbox_url = strtolower($res['inbox']);
-					$profile->public_key = $res['publicKey']['publicKeyPem'];
+						$profile->outbox_url = strtolower($res['outbox']);
-					$profile->key_id = $res['publicKey']['id'];
+						$profile->remote_url = strtolower($res['id']);
-					$profile->webfinger = strtolower(Purify::clean($webfinger));
+						$profile->public_key = $res['publicKey']['publicKeyPem'];
-					$profile->last_fetched_at = now();
+						$profile->key_id = $res['publicKey']['id'];
-					$profile->save();
+						$profile->webfinger = strtolower(Purify::clean($webfinger));
-					if(config('pixelfed.cloud_storage') == true) {
+						$profile->last_fetched_at = now();
-						RemoteAvatarFetch::dispatch($profile);
+						$profile->save();
-					}
+						if(config('pixelfed.cloud_storage') == true) {
-					return $profile;
+							RemoteAvatarFetch::dispatch($profile);
+						}
+						return $profile;
+					});
 				});
 			} else {
 				// Update info after 24 hours

routes/web.php

@@ -244,13 +244,13 @@ Route::domain(config('pixelfed.domain.app'))->middleware(['validemail', 'twofact
     Route::group(['prefix' => 'i'], function () {
         Route::redirect('/', '/');
         Route::get('compose', 'StatusController@compose')->name('compose');
-        Route::post('comment', 'CommentController@store')->middleware('throttle:maxCommentsPerHour,60')->middleware('throttle:maxCommentsPerDay,1440');
+        Route::post('comment', 'CommentController@store')->middleware('throttle:maxCommentsPerDay,1440');
         Route::post('delete', 'StatusController@delete');
         Route::post('mute', 'AccountController@mute');
         Route::post('unmute', 'AccountController@unmute');
         Route::post('block', 'AccountController@block');
         Route::post('unblock', 'AccountController@unblock');
-        Route::post('like', 'LikeController@store')->middleware('throttle:maxLikesPerHour,60')->middleware('throttle:maxLikesPerDay,1440');
+        Route::post('like', 'LikeController@store')->middleware('throttle:maxLikesPerDay,1440');
         Route::post('share', 'StatusController@storeShare')->middleware('throttle:maxSharesPerHour,60')->middleware('throttle:maxSharesPerDay,1440');
         Route::post('follow', 'FollowerController@store');
         Route::post('bookmark', 'BookmarkController@store');