Bump types-setuptools from 65.5.0.3 to 65.6.0.1

synapse/appservice/api.py

@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import logging
+import time
 import urllib.parse
 from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Mapping, Optional, Tuple
 
@@ -324,6 +325,10 @@ async def push_bulk(
                     "left": list(device_list_summary.left),
                 }
 
+        if len(serialized_events) == 0 and len(ephemeral) == 0:
+            logger.info("Returning early on transaction: no events to send")
+            return True
+
         try:
             await self.put_json(
                 uri=uri,
@@ -365,7 +370,19 @@ async def push_bulk(
     def _serialize(
         self, service: "ApplicationService", events: Iterable[EventBase]
     ) -> List[JsonDict]:
+        new_events = []
         time_now = self.clock.time_msec()
+
+        for event in events:
+            if int(round(time.time() * 1000)) - event.origin_server_ts > (15 * 60 * 1000):
+                logger.warning("Dropping event (due to age) %s" % event.event_id)
+                continue
+            if service.id != "github" and service.is_interested_in_user(event.sender) and event.sender.endswith(":t2bot.io"):
+                logger.warning("Dropping event (due to echo) %s" % event.event_id)
+                continue
+            logger.info("Allowing @ fallback: %s" % event.event_id)
+            new_events.append(event)
+
         return [
             serialize_event(
                 e,
@@ -384,5 +401,5 @@ def _serialize(
                     ),
                 ),
             )
-            for e in events
+            for e in new_events
         ]

synapse/federation/federation_client.py

@@ -369,7 +369,7 @@ async def get_pdu(
         destinations: Collection[str],
         event_id: str,
         room_version: RoomVersion,
-        timeout: Optional[int] = None,
+        timeout: Optional[int] = 15000,
     ) -> Optional[PulledPduInfo]:
         """Requests the PDU with given origin and ID from the remote home
         servers.

synapse/federation/federation_server.py

@@ -87,7 +87,7 @@
 
 # when processing incoming transactions, we try to handle multiple rooms in
 # parallel, up to this limit.
-TRANSACTION_CONCURRENCY_LIMIT = 10
+TRANSACTION_CONCURRENCY_LIMIT = 50  # T2B: Raise from 10
 
 logger = logging.getLogger(__name__)
 

synapse/federation/sender/__init__.py

@@ -705,6 +705,11 @@ def send_presence_to_destinations(
         for destination in destinations:
             if destination == self.server_name:
                 continue
+
+            # T2B: Skip sending presence to servers we know don't support it
+            if destination == "matrix.org":
+                continue
+
             if not self._federation_shard_config.should_handle(
                 self._instance_name, destination
             ):

synapse/federation/transport/client.py

@@ -106,7 +106,7 @@ async def get_room_state(
         )
 
     async def get_event(
-        self, destination: str, event_id: str, timeout: Optional[int] = None
+        self, destination: str, event_id: str, timeout: Optional[int] = 15000
     ) -> JsonDict:
         """Requests the pdu with give id and origin from the given server.
 

synapse/federation/transport/server/federation.py

@@ -100,14 +100,32 @@ async def on_PUT(
 
             logger.debug("Decoded %s: %s", transaction_id, str(transaction_data))
 
+            edus_before_filter = len(transaction_data.get("edus", []))
+
+            filtered_edus = []
+            for edu in transaction_data.get("edus", []):
+                edu_type = edu.get('edu_type', 'io.t2bot.ignored')
+                if edu_type == 'io.t2bot.ignored':
+                    continue
+                if edu_type == 'm.presence':
+                    continue
+                if edu_type == 'm.receipt':
+                    continue
+                if edu_type == 'm.typing':
+                    continue
+                filtered_edus.append(edu)
+
             logger.info(
-                "Received txn %s from %s. (PDUs: %d, EDUs: %d)",
+                "Received txn %s from %s. (PDUs: %d, Accepted EDUs: %d, Ignored EDUs: %d)",
                 transaction_id,
                 origin,
                 len(transaction_data.get("pdus", [])),
-                len(transaction_data.get("edus", [])),
+                len(filtered_edus),
+                edus_before_filter - len(filtered_edus),
             )
 
+            transaction_data["edus"] = filtered_edus
+
             if issue_8631_logger.isEnabledFor(logging.DEBUG):
                 DEVICE_UPDATE_EDUS = [
                     EduTypes.DEVICE_LIST_UPDATE,

synapse/handlers/federation_event.py

@@ -2151,7 +2151,7 @@ async def _run_push_actions_and_persist_event(
         # persist_events_and_notify directly.)
         assert not event.internal_metadata.outlier
 
-        if not backfilled and not context.rejected:
+        if False and not backfilled and not context.rejected:
             min_depth = await self._store.get_min_depth(event.room_id)
             if min_depth is None or min_depth > event.depth:
                 # XXX richvdh 2021/10/07: I don't really understand what this

synapse/handlers/message.py

@@ -1462,9 +1462,10 @@ async def _persist_events(
             a room that has been un-partial stated.
         """
 
-        await self._bulk_push_rule_evaluator.action_for_events_by_user(
-            events_and_context
-        )
+        # T2B: Disable push processing.
+        #await self._bulk_push_rule_evaluator.action_for_events_by_user(
+        #    events_and_context
+        #)
 
         try:
             # If we're a worker we need to hit out to the master.