This is an automated email from the ASF dual-hosted git repository. sebb pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/incubator-ponymail-foal.git
commit 16d25bef321682d0b88c24ccf61de7889b30fdb6 Author: Sebb <[email protected]> AuthorDate: Mon Dec 13 16:45:12 2021 +0000 Might as well handle the entire scan batch at once --- server/endpoints/mbox.py | 21 ++++++------- server/plugins/database.py | 5 ++-- server/plugins/messages.py | 75 +++++++++++++++++++++++++--------------------- 3 files changed, 54 insertions(+), 47 deletions(-) diff --git a/server/endpoints/mbox.py b/server/endpoints/mbox.py index 8b34362..624a889 100644 --- a/server/endpoints/mbox.py +++ b/server/endpoints/mbox.py @@ -103,21 +103,22 @@ async def process( response.enable_chunked_encoding() await response.prepare(request) - async for email in plugins.messages.query_each( + async for emails in plugins.messages.query_batch( session, query_defuzzed, metadata_only=True, epoch_order="asc" ): - mboxrd_source = await convert_source(session, email) - # Ensure each non-empty source ends with a blank line - if not mboxrd_source.endswith("\n\n"): - mboxrd_source += "\n" - try: - async with server.streamlock: - await asyncio.wait_for(response.write(mboxrd_source.encode("utf-8")), timeout=5) - except (TimeoutError, RuntimeError, CancelledError): - break # Writing stream failed, break it off. + for email in emails: + mboxrd_source = await convert_source(session, email) + # Ensure each non-empty source ends with a blank line + if not mboxrd_source.endswith("\n\n"): + mboxrd_source += "\n" + try: + async with server.streamlock: + await asyncio.wait_for(response.write(mboxrd_source.encode("utf-8")), timeout=5) + except (TimeoutError, RuntimeError, CancelledError): + break # Writing stream failed, break it off. return response diff --git a/server/plugins/database.py b/server/plugins/database.py index f322861..cf1f436 100644 --- a/server/plugins/database.py +++ b/server/plugins/database.py @@ -120,7 +120,7 @@ class Database: request_timeout=60, clear_scroll=True, scroll_kwargs=None, - **kwargs) -> typing.AsyncIterator[dict]: + **kwargs) -> typing.AsyncIterator[typing.List[dict]]: scroll_kwargs = scroll_kwargs or {} @@ -137,8 +137,7 @@ class Database: # While we can scroll, fetch a page try: while scroll_id and resp["hits"]["hits"]: - for hit in resp["hits"]["hits"]: - yield hit + yield resp["hits"]["hits"] resp = await self.client.scroll( body={"scroll_id": scroll_id, "scroll": scroll}, **scroll_kwargs ) diff --git a/server/plugins/messages.py b/server/plugins/messages.py index 47ca7d7..9c63458 100644 --- a/server/plugins/messages.py +++ b/server/plugins/messages.py @@ -316,7 +316,7 @@ async def get_source(session: plugins.session.SessionObject, permalink: str = No return None -async def query_each( +async def query_batch( session: plugins.session.SessionObject, query_defuzzed, hide_deleted=True, @@ -327,7 +327,7 @@ async def query_each( """ Advanced query and grab for stats.py Also called by mbox.py (using metadata_only=True) - Yields results singly + Yields batches of scan results, filtered to remove inaccessible mails """ assert session.database, DATABASE_NOT_CONNECTED preserve_order = True if epoch_order == "asc" else False @@ -345,38 +345,42 @@ async def query_each( es_query["_source"] = temp else: es_query["_source"] = { "excludes": ["body"] } - async for hit in session.database.scan( + async for hits in session.database.scan( query=es_query, preserve_order=preserve_order ): - doc = hit["_source"] - # If email was delete/hidden and we're not doing an admin query, ignore it - if hide_deleted and doc.get("deleted", False): - continue - if plugins.aaa.can_access_email(session, doc): - if "mid" in doc: # might be missing when using source_fields - doc["id"] = doc["mid"] - # Calculate gravatars if not present in source - if not metadata_only and source_fields is None and "gravatar" not in doc: - doc["gravatar"] = gravatar(doc) - if not session.credentials: - doc = anonymize(doc) - if "body_short" in doc: - # The body_short field is set to SHORT_BODY_MAX_LEN+1 if the body is longer - # than SHORT_BODY_MAX_LEN, so we know if it has been truncated - if len(doc["body_short"] or "") > SHORT_BODY_MAX_LEN: - doc["body"] = doc["body_short"][:SHORT_BODY_MAX_LEN] + '...' - else: - doc["body"] = doc["body_short"] - # stats.py is expecting doc['body'], not body_short - del doc["body_short"] - trim_email(doc) - # drop any added fields - if not source_fields is None: - for hdr in MUST_HAVE: - if not hdr in source_fields and hdr in doc: - del doc[hdr] - yield doc + docs = [] + for hit in hits: + doc = hit["_source"] + # If email was delete/hidden and we're not doing an admin query, ignore it + if hide_deleted and doc.get("deleted", False): + continue + if plugins.aaa.can_access_email(session, doc): + if "mid" in doc: # might be missing when using source_fields + doc["id"] = doc["mid"] + # Calculate gravatars if not present in source + if not metadata_only and source_fields is None and "gravatar" not in doc: + doc["gravatar"] = gravatar(doc) + if not session.credentials: + doc = anonymize(doc) + if "body_short" in doc: + # The body_short field is set to SHORT_BODY_MAX_LEN+1 if the body is longer + # than SHORT_BODY_MAX_LEN, so we know if it has been truncated + if len(doc["body_short"] or "") > SHORT_BODY_MAX_LEN: + doc["body"] = doc["body_short"][:SHORT_BODY_MAX_LEN] + '...' + else: + doc["body"] = doc["body_short"] + # stats.py is expecting doc['body'], not body_short + del doc["body_short"] + trim_email(doc) + # drop any added fields + if not source_fields is None: + for hdr in MUST_HAVE: + if not hdr in source_fields and hdr in doc: + del doc[hdr] + docs.append(doc) + if len(docs) > 0: + yield docs async def query( @@ -394,7 +398,7 @@ async def query( """ docs = [] hits = 0 - async for doc in query_each( + async for batch in query_batch( session, query_defuzzed, hide_deleted=hide_deleted, @@ -402,8 +406,11 @@ async def query( epoch_order=epoch_order, source_fields=source_fields ): - docs.append(doc) - hits += 1 + for doc in batch: + docs.append(doc) + hits += 1 + if hits > query_limit: + break if hits > query_limit: break return docs
