Core/Provisioner/Provisioners/cassandra-provision.cql

@@ -1379,7 +1379,7 @@ CREATE TABLE minds.analytics_graphs (
  last_synced timestamp,
  data text,
  PRIMARY KEY(key)
-); 
+);
 
 CREATE TABLE minds.views (
  year int,
@@ -1415,6 +1415,13 @@ CREATE TABLE minds.hidden_hashtags (
  PRIMARY KEY (hashtag)
 );
 
+CREATE TABLE minds.search_dispatcher_queue (
+ entity_urn text,
+ last_retry timestamp,
+ retries int,
+ PRIMARY KEY (entity_urn)
+);
+
 CREATE TABLE minds.user_snapshots (
  user_guid varint,
  type text,

Core/Search/Events.php

@@ -4,10 +4,10 @@
  */
 namespace Minds\Core\Search;
 
+use Exception;
 use Minds\Core;
 use Minds\Core\Di\Di;
 use Minds\Core\Events\Event;
-use Minds\Entities;
 
 class Events
 {
@@ -79,12 +79,30 @@ class Events
  return;
  }
 
- Di::_()->get('Search\Index')
- ->index(
- is_string($params['entity']) ?
- unserialize($params['entity']) :
- $params['entity']
- );
+ $entity = is_string($params['entity']) ?
+ unserialize($params['entity']) :
+ $params['entity'];
+
+ try {
+ $wasIndexed = (bool) Di::_()->get('Search\Index')
+ ->index($entity);
+ } catch (Exception $e) {
+ error_log("[Search/Events/search:index:dispatch] {$e}");
+ $wasIndexed = false;
+ }
+
+ // BannedException will return null (which is also falsy)
+ // So we should retry only on non-null responses from index()
+ if ($wasIndexed !== null) {
+ /** @var Core\Search\RetryQueue\Manager $retryQueueManager */
+ $retryQueueManager = Di::_()->get('Search\RetryQueue\Manager');
+
+ if ($wasIndexed) {
+ $retryQueueManager->prune($entity);
+ } else {
+ $retryQueueManager->retry($entity);
+ }
+ }
 
  } catch (\Exception $e) {
  error_log('[Search/Events/search:index:dispatch] ' . get_class($e) . ': ' . $e->getMessage());

Core/Search/Index.php

@@ -93,7 +93,7 @@ class Index
  }
  }
  } catch (BannedException $e) {
- $result = false;
+ $result = null;
  } catch (\Exception $e) {
  error_log('[Search/Index] ' . get_class($e) . ": {$e->getMessage()}");
  $result = false;

Core/Search/RetryQueue/Manager.php

+<?php
+/**
+ * RetryQueueManager
+ * @author edgebal
+ */
+
+namespace Minds\Core\Search\RetryQueue;
+
+use Exception;
+use Minds\Common\Urn;
+use Minds\Core\Di\Di;
+use Minds\Core\Events\EventsDispatcher;
+
+class Manager
+{
+ /** @var EventsDispatcher */
+ protected $eventsDispatcher;
+
+ /** @var Repository */
+ protected $repository;
+
+ /**
+ * RetryQueueManager constructor.
+ * @param EventsDispatcher $eventsDispatcher
+ * @param Repository $repository
+ */
+ public function __construct(
+ $eventsDispatcher = null,
+ $repository = null
+ )
+ {
+ $this->eventsDispatcher = $eventsDispatcher ?: Di::_()->get('EventsDispatcher');
+ $this->repository = $repository ?: new Repository();
+ }
+
+ /**
+ * @param mixed $entity
+ * @return bool
+ * @throws Exception
+ */
+ public function prune($entity)
+ {
+ $urn = (string) (new Urn($entity->guid));
+
+ $retryQueueEntry = new RetryQueueEntry();
+ $retryQueueEntry
+ ->setEntityUrn($urn);
+
+ return (bool) $this->repository->delete($retryQueueEntry);
+ }
+
+ /**
+ * @param mixed $entity
+ * @return bool
+ * @throws Exception
+ */
+ public function retry($entity)
+ {
+ $urn = (string) (new Urn($entity->guid));
+
+ $retryQueueEntry = $this->repository->get($urn);
+ $retries = $retryQueueEntry->getRetries() + 1;
+
+ $retryQueueEntry
+ ->setLastRetry(time())
+ ->setRetries($retries);
+
+ $retrySaved = $this->repository->add($retryQueueEntry);
+
+ if (!$retrySaved) {
+ error_log("[RetryQueueManager] Critical: Cannot save retry to queue table: {$urn}");
+ } elseif ($retries < 5) {
+ error_log("[RetryQueueManager] Warn: Re-queue: {$urn}");
+
+ $this->eventsDispatcher->trigger('search:index', 'all', [
+ 'entity' => $entity
+ ]);
+ } else {
+ error_log("[RetryQueueManager] Critical: Too many retries indexing: {$urn}");
+ }
+
+ return $retrySaved;
+ }
+}

Core/Search/RetryQueue/Repository.php

+<?php
+/**
+ * Repository
+ * @author edgebal
+ */
+
+namespace Minds\Core\Search\RetryQueue;
+
+use Cassandra\Rows;
+use Cassandra\Timestamp;
+use Exception;
+use Minds\Common\Repository\Response;
+use Minds\Core\Data\Cassandra\Client as CassandraClient;
+use Minds\Core\Data\Cassandra\Prepared\Custom;
+use Minds\Core\Di\Di;
+
+class Repository
+{
+ /** @var CassandraClient */
+ protected $db;
+
+ public function __construct(
+ $db = null
+ )
+ {
+ $this->db = $db ?: Di::_()->get('Database\Cassandra\Cql');
+ }
+
+ /**
+ * @param array $opts
+ * @return Response
+ */
+ public function getList(array $opts = [])
+ {
+ $opts = array_merge([
+ 'entity_urn' => null,
+ ], $opts);
+
+ $cql = "SELECT * FROM search_dispatcher_queue";
+ $values = [];
+ $cqlOpts = [];
+
+ if ($opts['entity_urn']) {
+ $cql .= ' WHERE entity_urn = ?';
+ $values[] = $opts['entity_urn'];
+ }
+
+ if ($opts['limit'] ?? null) {
+ $cqlOpts['page_size'] = (int) $opts['limit'];
+ }
+
+ if ($opts['offset'] ?? null) {
+ $cqlOpts['paging_state_token'] = base64_decode($opts['offset']);
+ }
+
+ $prepared = new Custom();
+ $prepared->query($cql, $values);
+ $prepared->setOpts($cqlOpts);
+
+ $response = new Response();
+
+ try {
+ /** @var Rows $rows */
+ $rows = $this->db->request($prepared);
+
+ if ($rows) {
+ foreach ($rows as $row) {
+ $retryQueueEntry = new RetryQueueEntry();
+ $retryQueueEntry
+ ->setEntityUrn($row['entity_urn'])
+ ->setLastRetry($row['last_retry']->time())
+ ->setRetries($row['retries']);
+
+ $response[] = $retryQueueEntry;
+ }
+
+ $response->setPagingToken(base64_encode($rows->pagingStateToken()));
+ $response->setLastPage($rows->isLastPage());
+ }
+ } catch (Exception $e) {
+ error_log($e);
+ $response->setException($e);
+ }
+
+ return $response;
+ }
+
+ /**
+ * @param $urn
+ * @return RetryQueueEntry
+ */
+ public function get($urn)
+ {
+ $retryQueueEntries = $this->getList([
+ 'entity_urn' => $urn,
+ ])->toArray();
+
+ if (count($retryQueueEntries)) {
+ return $retryQueueEntries[0];
+ } else {
+ $retryQueueEntry = new RetryQueueEntry();
+ $retryQueueEntry
+ ->setEntityUrn($urn)
+ ->setLastRetry(time())
+ ->setRetries(0);
+
+ return $retryQueueEntry;
+ }
+ }
+
+ /**
+ * @param RetryQueueEntry $retryQueueEntry
+ * @return bool
+ * @throws Exception
+ */
+ public function add(RetryQueueEntry $retryQueueEntry)
+ {
+ if (!$retryQueueEntry->getEntityUrn()) {
+ throw new Exception('Missing URN');
+ }
+
+ $cql = "INSERT INTO search_dispatcher_queue (entity_urn, last_retry, retries) VALUES (?, ?, ?)";
+ $values = [
+ (string) $retryQueueEntry->getEntityUrn(),
+ new Timestamp($retryQueueEntry->getLastRetry()),
+ (int) $retryQueueEntry->getRetries(),
+ ];
+
+ $prepared = new Custom();
+ $prepared->query($cql, $values);
+
+ try {
+ return (bool) $this->db->request($prepared, true);
+ } catch (Exception $e) {
+ error_log($e);
+ return false;
+ }
+ }
+
+ /**
+ * @param RetryQueueEntry $retryQueueEntry
+ * @return bool
+ * @throws Exception
+ */
+ public function update(RetryQueueEntry $retryQueueEntry)
+ {
+ return $this->add($retryQueueEntry);
+ }
+
+ /**
+ * @param RetryQueueEntry $retryQueueEntry
+ * @return bool
+ * @throws Exception
+ */
+ public function delete(RetryQueueEntry $retryQueueEntry)
+ {
+ if (!$retryQueueEntry->getEntityUrn()) {
+ throw new Exception('Missing URN');
+ }
+
+ $cql = "DELETE FROM search_dispatcher_queue WHERE entity_urn = ?";
+ $values = [
+ (string) $retryQueueEntry->getEntityUrn(),
+ ];
+
+ $prepared = new Custom();
+ $prepared->query($cql, $values);
+
+ try {
+ return (bool) $this->db->request($prepared, true);
+ } catch (Exception $e) {
+ error_log($e);
+ return false;
+ }
+ }
+}

Core/Search/RetryQueue/RetryQueueEntry.php

+<?php
+/**
+ * RetryQueueEntry
+ * @author edgebal
+ */
+
+namespace Minds\Core\Search\RetryQueue;
+
+use Minds\Traits\MagicAttributes;
+
+/**
+ * Class RetryQueueEntry
+ * @package Minds\Core\Search\RetryQueue
+ * @method string getEntityUrn()
+ * @method RetryQueueEntry setEntityUrn(string $entityUrn)
+ * @method int getRetries()
+ * @method RetryQueueEntry setRetries(int $retries)
+ * @method int getLastRetry()
+ * @method RetryQueueEntry setLastRetry(int $entityUrn)
+ */
+class RetryQueueEntry
+{
+ use MagicAttributes;
+
+ protected $entityUrn;
+
+ protected $retries;
+
+ protected $lastRetry;
+}

Core/Search/SearchProvider.php

@@ -9,7 +9,6 @@
 namespace Minds\Core\Search;
 
 use Minds\Core\Di\Provider;
-use Minds\Core\Search\Hashtags\Manager;
 
 class SearchProvider extends Provider
 {
@@ -40,7 +39,11 @@ class SearchProvider extends Provider
  }, ['useFactory' => true]);
 
  $this->di->bind('Search\Hashtags\Manager', function ($di) {
- return new Manager();
+ return new Hashtags\Manager();
+ }, ['useFactory' => true]);
+
+ $this->di->bind('Search\RetryQueue\Manager', function ($di) {
+ return new RetryQueue\Manager();
  }, ['useFactory' => true]);
  }
 }

Spec/Core/Search/RetryQueue/ManagerSpec.php

+<?php
+
+namespace Spec\Minds\Core\Search\RetryQueue;
+
+use Minds\Core\Events\EventsDispatcher;
+use Minds\Core\Search\RetryQueue\Manager;
+use Minds\Core\Search\RetryQueue\Repository;
+use Minds\Core\Search\RetryQueue\RetryQueueEntry;
+use Minds\Entities\Entity;
+use PhpSpec\ObjectBehavior;
+use Prophecy\Argument;
+
+class ManagerSpec extends ObjectBehavior
+{
+ /** @var EventsDispatcher */
+ protected $eventsDispatcher;
+
+ /** @var Repository */
+ protected $repository;
+
+ function let(
+ EventsDispatcher $eventsDispatcher,
+ Repository $repository
+ )
+ {
+ $this->beConstructedWith($eventsDispatcher, $repository);
+
+ $this->eventsDispatcher = $eventsDispatcher;
+ $this->repository = $repository;
+ }
+
+ function it_is_initializable()
+ {
+ $this->shouldHaveType(Manager::class);
+ }
+
+ function it_should_prune(Entity $entity)
+ {
+ $entity->get('guid')
+ ->shouldBeCalled()
+ ->willReturn('5000');
+
+ $this->repository->delete(Argument::type(RetryQueueEntry::class))
+ ->shouldBeCalled()
+ ->willReturn(true);
+
+ $this
+ ->prune($entity)
+ ->shouldReturn(true);
+ }
+
+ function it_should_retry(Entity $entity, RetryQueueEntry $retryQueueEntry)
+ {
+ $entity->get('guid')
+ ->shouldBeCalled()
+ ->willReturn('5000');
+
+ $this->repository->get('urn:entity:5000')
+ ->shouldBeCalled()
+ ->willReturn($retryQueueEntry);
+
+ $retryQueueEntry->getRetries()
+ ->shouldBeCalled()
+ ->willReturn(3);
+
+ $retryQueueEntry->setLastRetry(Argument::type('int'))
+ ->shouldBeCalled()
+ ->willReturn($retryQueueEntry);
+
+ $retryQueueEntry->setRetries(4)
+ ->shouldBeCalled()
+ ->willReturn($retryQueueEntry);
+
+ $this->repository->add($retryQueueEntry)
+ ->shouldBeCalled()
+ ->willReturn(true);
+
+ $this->eventsDispatcher->trigger('search:index', 'all', [
+ 'entity' => $entity
+ ])
+ ->shouldBeCalled()
+ ->willReturn(true);
+
+ $this
+ ->retry($entity)
+ ->shouldReturn(true);
+ }
+
+ function it_should_not_retry_if_too_many_attempts(Entity $entity, RetryQueueEntry $retryQueueEntry)
+ {
+ $entity->get('guid')
+ ->shouldBeCalled()
+ ->willReturn('5000');
+
+ $this->repository->get('urn:entity:5000')
+ ->shouldBeCalled()
+ ->willReturn($retryQueueEntry);
+
+ $retryQueueEntry->getRetries()
+ ->shouldBeCalled()
+ ->willReturn(4);
+
+ $retryQueueEntry->setLastRetry(Argument::type('int'))
+ ->shouldBeCalled()
+ ->willReturn($retryQueueEntry);
+
+ $retryQueueEntry->setRetries(5)
+ ->shouldBeCalled()
+ ->willReturn($retryQueueEntry);
+
+ $this->repository->add($retryQueueEntry)
+ ->shouldBeCalled()
+ ->willReturn(true);
+
+ $this->eventsDispatcher->trigger('search:index', Argument::cetera())
+ ->shouldNotBeCalled();
+
+ $this
+ ->retry($entity)
+ ->shouldReturn(true);
+ }
+}

Spec/Core/Search/RetryQueue/RepositorySpec.php

+<?php
+
+namespace Spec\Minds\Core\Search\RetryQueue;
+
+use Exception;
+use Minds\Common\Repository\Response;
+use Minds\Core\Data\Cassandra\Client as CassandraClient;
+use Minds\Core\Data\Cassandra\Prepared\Custom;
+use Minds\Core\Search\RetryQueue\Repository;
+use Minds\Core\Search\RetryQueue\RetryQueueEntry;
+use PhpSpec\ObjectBehavior;
+use Prophecy\Argument;
+use Spec\Minds\Mocks\Cassandra\Rows;
+
+class RepositorySpec extends ObjectBehavior
+{
+ /** @var CassandraClient */
+ protected $db;
+
+ function let(
+ CassandraClient $db
+ )
+ {
+ $this->beConstructedWith($db);
+
+ $this->db = $db;
+ }
+
+ function it_is_initializable()
+ {
+ $this->shouldHaveType(Repository::class);
+ }
+
+ function it_should_get_a_list()
+ {
+ $this->db->request(Argument::that(function (Custom $prepared) {
+ $query = $prepared->build();
+
+ return $query['string'] == 'SELECT * FROM search_dispatcher_queue' &&
+ $query['values'] === [];
+ }))
+ ->shouldBeCalled()
+ ->willReturn(new Rows([], '1a2b3c4d5e6f7890'));
+
+ $this
+ ->getList([])
+ ->shouldReturnAnInstanceOf(Response::class);
+ }
+
+ function it_should_get_by_a_list_by_urn()
+ {
+ $this->db->request(Argument::that(function (Custom $prepared) {
+ $query = $prepared->build();
+
+ return $query['string'] == 'SELECT * FROM search_dispatcher_queue WHERE entity_urn = ?' &&
+ $query['values'] === ['urn:test:123456'];
+ }))
+ ->shouldBeCalled()
+ ->willReturn(new Rows([
+ [
+ 'entity_urn' => 'urn:test:123456',
+ 'last_retry' => new \Cassandra\Timestamp(1562182542),
+ 'retries' => 3,
+ ]
+ ], '1a2b3c4d5e6f7890'));
+
+ $this
+ ->getList([
+ 'entity_urn' => 'urn:test:123456',
+ ])
+ ->shouldReturnAnInstanceOf(Response::class);
+ }
+
+ function it_should_add(RetryQueueEntry $retryQueueEntry)
+ {
+ $retryQueueEntry->getEntityUrn()
+ ->shouldBeCalled()
+ ->willReturn('urn:test:123456');
+
+ $retryQueueEntry->getLastRetry()
+ ->shouldBeCalled()
+ ->willReturn(1562182542);
+
+ $retryQueueEntry->getRetries()
+ ->shouldBeCalled()
+ ->willReturn(3);
+
+ $this->db->request(Argument::that(function (Custom $prepared) {
+ $query = $prepared->build();
+
+ return stripos($query['string'], 'INSERT INTO search_dispatcher_queue') === 0;
+ }), true)
+ ->shouldBeCalled()
+ ->willReturn(true);
+
+ $this
+ ->add($retryQueueEntry)
+ ->shouldNotReturn(false);
+ }
+
+ function it_should_throw_during_add_if_no_entity_urn(RetryQueueEntry $retryQueueEntry)
+ {
+ $retryQueueEntry->getEntityUrn()
+ ->shouldBeCalled()
+ ->willReturn(null);
+
+ $this->db->request(Argument::cetera())
+ ->shouldNotBeCalled();
+
+ $this
+ ->shouldThrow(new Exception('Missing URN'))
+ ->duringAdd($retryQueueEntry);
+ }
+
+ function it_should_delete(RetryQueueEntry $retryQueueEntry)
+ {
+ $retryQueueEntry->getEntityUrn()
+ ->shouldBeCalled()
+ ->willReturn('urn:test:123456');
+
+ $this->db->request(Argument::that(function (Custom $prepared) {
+ $query = $prepared->build();
+
+ return stripos($query['string'], 'DELETE FROM search_dispatcher_queue') === 0;
+ }), true)
+ ->shouldBeCalled()
+ ->willReturn(true);
+
+ $this
+ ->delete($retryQueueEntry)
+ ->shouldNotReturn(false);
+ }
+
+ function it_should_throw_during_delete_if_no_entity_urn(RetryQueueEntry $retryQueueEntry)
+ {
+ $retryQueueEntry->getEntityUrn()
+ ->shouldBeCalled()
+ ->willReturn(null);
+
+ $this->db->request(Argument::cetera())
+ ->shouldNotBeCalled();
+
+ $this
+ ->shouldThrow(new Exception('Missing URN'))
+ ->duringDelete($retryQueueEntry);
+ }
+}