Merge branch 'master' into fix-error-message-in-delivery-worker

Gemfile

@@ -148,3 +148,4 @@ group :production do
 end
 
 gem 'concurrent-ruby', require: false
+gem 'connection_pool', require: false

Gemfile.lock

@@ -666,6 +666,7 @@ DEPENDENCIES
   cld3 (~> 3.2.4)
   climate_control (~> 0.2)
   concurrent-ruby
+  connection_pool
   derailed_benchmarks
   devise (~> 4.6)
   devise-two-factor (~> 3.0)

app/javascript/mastodon/features/status/index.js

@@ -4,6 +4,7 @@ import { connect } from 'react-redux';
 import PropTypes from 'prop-types';
 import classNames from 'classnames';
 import ImmutablePropTypes from 'react-immutable-proptypes';
+import { createSelector } from 'reselect';
 import { fetchStatus } from '../../actions/statuses';
 import MissingIndicator from '../../components/missing_indicator';
 import DetailedStatus from './components/detailed_status';
@@ -63,39 +64,58 @@ const messages = defineMessages({
 const makeMapStateToProps = () => {
   const getStatus = makeGetStatus();
 
-  const mapStateToProps = (state, props) => {
-    const status = getStatus(state, { id: props.params.statusId });
+  const getAncestorsIds = createSelector([
+    (_, { id }) => id,
+    state => state.getIn(['contexts', 'inReplyTos']),
+  ], (statusId, inReplyTos) => {
     let ancestorsIds = Immutable.List();
+    ancestorsIds = ancestorsIds.withMutations(mutable => {
+      let id = statusId;
+
+      while (id) {
+        mutable.unshift(id);
+        id = inReplyTos.get(id);
+      }
+    });
+
+    return ancestorsIds;
+  });
+
+  const getDescendantsIds = createSelector([
+    (_, { id }) => id,
+    state => state.getIn(['contexts', 'replies']),
+  ], (statusId, contextReplies) => {
     let descendantsIds = Immutable.List();
+    descendantsIds = descendantsIds.withMutations(mutable => {
+      const ids = [statusId];
 
-    if (status) {
-      ancestorsIds = ancestorsIds.withMutations(mutable => {
-        let id = status.get('in_reply_to_id');
+      while (ids.length > 0) {
+        let id        = ids.shift();
+        const replies = contextReplies.get(id);
 
-        while (id) {
-          mutable.unshift(id);
-          id = state.getIn(['contexts', 'inReplyTos', id]);
+        if (statusId !== id) {
+          mutable.push(id);
         }
-      });
 
-      descendantsIds = descendantsIds.withMutations(mutable => {
-        const ids = [status.get('id')];
+        if (replies) {
+          replies.reverse().forEach(reply => {
+            ids.unshift(reply);
+          });
+        }
+      }
+    });
 
-        while (ids.length > 0) {
-          let id        = ids.shift();
-          const replies = state.getIn(['contexts', 'replies', id]);
+    return descendantsIds;
+  });
 
-          if (status.get('id') !== id) {
-            mutable.push(id);
-          }
+  const mapStateToProps = (state, props) => {
+    const status = getStatus(state, { id: props.params.statusId });
+    let ancestorsIds = Immutable.List();
+    let descendantsIds = Immutable.List();
 
-          if (replies) {
-            replies.reverse().forEach(reply => {
-              ids.unshift(reply);
-            });
-          }
-        }
-      });
+    if (status) {
+      ancestorsIds = getAncestorsIds(state, { id: status.get('in_reply_to_id') });
+      descendantsIds = getDescendantsIds(state, { id: status.get('id') });
     }
 
     return {

app/javascript/mastodon/reducers/compose.js

@@ -195,6 +195,12 @@ const expandMentions = status => {
   return fragment.innerHTML;
 };
 
+const expiresInFromExpiresAt = expires_at => {
+  if (!expires_at) return 24 * 3600;
+  const delta = (new Date(expires_at).getTime() - Date.now()) / 1000;
+  return [300, 1800, 3600, 21600, 86400, 259200, 604800].find(expires_in => expires_in >= delta) || 24 * 3600;
+};
+
 export default function compose(state = initialState, action) {
   switch(action.type) {
   case STORE_HYDRATE:
@@ -353,7 +359,7 @@ export default function compose(state = initialState, action) {
         map.set('poll', ImmutableMap({
           options: action.status.getIn(['poll', 'options']).map(x => x.get('title')),
           multiple: action.status.getIn(['poll', 'multiple']),
-          expires_in: 24 * 3600,
+          expires_in: expiresInFromExpiresAt(action.status.getIn(['poll', 'expires_at'])),
         }));
       }
     });

app/lib/connection_pool/shared_connection_pool.rb

@@ -0,0 +1,63 @@
+# frozen_string_literal: true
+
+require 'connection_pool'
+require_relative './shared_timed_stack'
+
+class ConnectionPool::SharedConnectionPool < ConnectionPool
+  def initialize(options = {}, &block)
+    super(options, &block)
+
+    @available = ConnectionPool::SharedTimedStack.new(@size, &block)
+  end
+
+  delegate :size, :flush, to: :@available
+
+  def with(preferred_tag, options = {})
+    Thread.handle_interrupt(Exception => :never) do
+      conn = checkout(preferred_tag, options)
+
+      begin
+        Thread.handle_interrupt(Exception => :immediate) do
+          yield conn
+        end
+      ensure
+        checkin(preferred_tag)
+      end
+    end
+  end
+
+  def checkout(preferred_tag, options = {})
+    if ::Thread.current[key(preferred_tag)]
+      ::Thread.current[key_count(preferred_tag)] += 1
+      ::Thread.current[key(preferred_tag)]
+    else
+      ::Thread.current[key_count(preferred_tag)] = 1
+      ::Thread.current[key(preferred_tag)] = @available.pop(preferred_tag, options[:timeout] || @timeout)
+    end
+  end
+
+  def checkin(preferred_tag)
+    if ::Thread.current[key(preferred_tag)]
+      if ::Thread.current[key_count(preferred_tag)] == 1
+        @available.push(::Thread.current[key(preferred_tag)])
+        ::Thread.current[key(preferred_tag)] = nil
+      else
+        ::Thread.current[key_count(preferred_tag)] -= 1
+      end
+    else
+      raise ConnectionPool::Error, 'no connections are checked out'
+    end
+
+    nil
+  end
+
+  private
+
+  def key(tag)
+    :"#{@key}-#{tag}"
+  end
+
+  def key_count(tag)
+    :"#{@key_count}-#{tag}"
+  end
+end

app/lib/connection_pool/shared_timed_stack.rb

@@ -0,0 +1,95 @@
+# frozen_string_literal: true
+
+class ConnectionPool::SharedTimedStack
+  def initialize(max = 0, &block)
+    @create_block = block
+    @max          = max
+    @created      = 0
+    @queue        = []
+    @tagged_queue = Hash.new { |hash, key| hash[key] = [] }
+    @mutex        = Mutex.new
+    @resource     = ConditionVariable.new
+  end
+
+  def push(connection)
+    @mutex.synchronize do
+      store_connection(connection)
+      @resource.broadcast
+    end
+  end
+
+  alias << push
+
+  def pop(preferred_tag, timeout = 5.0)
+    deadline = current_time + timeout
+
+    @mutex.synchronize do
+      loop do
+        return fetch_preferred_connection(preferred_tag) unless @tagged_queue[preferred_tag].empty?
+
+        connection = try_create(preferred_tag)
+        return connection if connection
+
+        to_wait = deadline - current_time
+        raise Timeout::Error, "Waited #{timeout} sec" if to_wait <= 0
+
+        @resource.wait(@mutex, to_wait)
+      end
+    end
+  end
+
+  def empty?
+    size.zero?
+  end
+
+  def size
+    @mutex.synchronize do
+      @queue.size
+    end
+  end
+
+  def flush
+    @mutex.synchronize do
+      @queue.delete_if do |connection|
+        delete = !connection.in_use && (connection.dead || connection.seconds_idle >= RequestPool::MAX_IDLE_TIME)
+
+        if delete
+          @tagged_queue[connection.site].delete(connection)
+          connection.close
+          @created -= 1
+        end
+
+        delete
+      end
+    end
+  end
+
+  private
+
+  def try_create(preferred_tag)
+    if @created == @max && !@queue.empty?
+      throw_away_connection = @queue.pop
+      @tagged_queue[throw_away_connection.site].delete(throw_away_connection)
+      @create_block.call(preferred_tag)
+    elsif @created != @max
+      connection = @create_block.call(preferred_tag)
+      @created += 1
+      connection
+    end
+  end
+
+  def fetch_preferred_connection(preferred_tag)
+    connection = @tagged_queue[preferred_tag].pop
+    @queue.delete(connection)
+    connection
+  end
+
+  def current_time
+    Process.clock_gettime(Process::CLOCK_MONOTONIC)
+  end
+
+  def store_connection(connection)
+    @tagged_queue[connection.site].push(connection)
+    @queue.push(connection)
+  end
+end