Skip to content

Commit 694d244

Browse files
themaroltthemarolt
committed
refactor: convert ListWorker class to plain functions (CM-1318)
Signed-off-by: Uroš Marolt <uros@marolt.me>
1 parent e993187 commit 694d244

2 files changed

Lines changed: 130 additions & 139 deletions

File tree

‎services/apps/mailing_list_integration/src/crowdmail/server.py‎

Lines changed: 4 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
from loguru import logger
77

88
from crowdmail.settings import WORKER_SHUTDOWN_TIMEOUT_SEC
9-
from crowdmail.worker.list_worker import ListWorker
9+
from crowdmail.worker.list_worker import run_worker, shutdown_worker
1010

1111

1212
@asynccontextmanager
@@ -15,15 +15,13 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
1515
logger.info("Starting application lifespan")
1616

1717
worker_task = None
18-
worker = ListWorker()
19-
logger.info("List worker initialized")
2018
try:
21-
worker_task = asyncio.create_task(worker.run())
22-
logger.info("ListWorker started successfully")
19+
worker_task = asyncio.create_task(run_worker())
20+
logger.info("List worker started successfully")
2321
# Yield control to FastAPI - it will handle signals and trigger shutdown
2422
yield
2523
finally:
26-
await worker.shutdown() # stopping worker to process new lists
24+
await shutdown_worker() # stopping worker to process new lists
2725
if worker_task is not None:
2826
try:
2927
await asyncio.wait_for(worker_task, timeout=WORKER_SHUTDOWN_TIMEOUT_SEC)

‎services/apps/mailing_list_integration/src/crowdmail/worker/list_worker.py‎

Lines changed: 126 additions & 133 deletions
Original file line numberDiff line numberDiff line change
@@ -27,142 +27,135 @@
2727
WORKER_POLLING_INTERVAL_SEC,
2828
)
2929

30-
31-
class ListWorker:
32-
"""Worker that mirrors, parses and emits activities for mailing lists"""
33-
34-
def __init__(self):
35-
self._shutdown = False
36-
37-
async def run(self):
38-
logger.info("Starting mailing list worker")
39-
try:
40-
await self._run()
41-
logger.info("Worker _run() method completed")
42-
finally:
43-
logger.info("Worker run() method exiting")
44-
45-
async def _run(self):
46-
try:
47-
logger.info("Starting list worker loop...")
48-
while not self._shutdown:
30+
_shutdown_event = asyncio.Event()
31+
32+
33+
async def run_worker():
34+
"""Worker loop that mirrors, parses and emits activities for mailing lists"""
35+
logger.info("Starting mailing list worker")
36+
try:
37+
logger.info("Starting list worker loop...")
38+
while not _shutdown_event.is_set():
39+
try:
40+
await _process_lists()
41+
await asyncio.sleep(WORKER_POLLING_INTERVAL_SEC)
42+
except Exception as e:
43+
logger.error("Worker error: {}", e)
44+
await asyncio.sleep(WORKER_ERROR_BACKOFF_SEC)
45+
logger.info("Worker loop completed")
46+
finally:
47+
await queue_service.shutdown()
48+
logger.info("Worker processing loop completed")
49+
50+
51+
async def shutdown_worker():
52+
logger.info("Shutting down list worker")
53+
_shutdown_event.set()
54+
55+
56+
async def _process_lists():
57+
mailing_list = None
58+
try:
59+
mailing_list = await acquire_list_for_processing()
60+
if not mailing_list:
61+
logger.debug("No mailing lists to process")
62+
return
63+
await _process_single_list(mailing_list)
64+
except Exception as e:
65+
logger.error(f"Failed to process mailing list {mailing_list} with error {e}")
66+
finally:
67+
if mailing_list:
68+
logger.info(f"releasing list: {mailing_list.source_url}")
69+
await release_list(mailing_list.id)
70+
logger.info(f"List {mailing_list.source_url} released!")
71+
72+
73+
async def _process_single_list(mailing_list: MailingList):
74+
logger.info("Processing mailing list: {}", mailing_list.source_url)
75+
state = ListState.FAILED
76+
77+
try:
78+
list_dir = await ensure_mirror(mailing_list.name, mailing_list.source_url)
79+
heads = dict(mailing_list.last_processed_heads)
80+
activities_db = []
81+
activities_kafka = []
82+
83+
for shard_path in discover_shards(list_dir):
84+
shard = shard_index(shard_path)
85+
commit_ids = await new_commits(shard_path, heads.get(shard))
86+
for git_id in commit_ids:
87+
heads[shard] = git_id
4988
try:
50-
await self._process_lists()
51-
await asyncio.sleep(WORKER_POLLING_INTERVAL_SEC)
89+
message, blob_id = await asyncio.to_thread(read_email, shard_path, git_id)
90+
parsed = parse_email(
91+
message,
92+
mailing_list.source_url,
93+
mailing_list.name,
94+
git_id,
95+
blob_id,
96+
mailing_list.segment_id,
97+
mailing_list.integration_id,
98+
)
5299
except Exception as e:
53-
logger.error("Worker error: {}", e)
54-
await asyncio.sleep(WORKER_ERROR_BACKOFF_SEC)
55-
logger.info("Worker loop completed")
56-
finally:
57-
await queue_service.shutdown()
58-
logger.info("Worker processing loop completed")
59-
60-
async def shutdown(self):
61-
logger.info("Shutting down list worker")
62-
self._shutdown = True
63-
64-
async def _process_lists(self):
65-
mailing_list = None
66-
try:
67-
mailing_list = await acquire_list_for_processing()
68-
if not mailing_list:
69-
logger.debug("No mailing lists to process")
70-
return
71-
await self._process_single_list(mailing_list)
72-
except Exception as e:
73-
logger.error(f"Failed to process mailing list {mailing_list} with error {e}")
74-
finally:
75-
if mailing_list:
76-
logger.info(f"releasing list: {mailing_list.source_url}")
77-
await release_list(mailing_list.id)
78-
logger.info(f"List {mailing_list.source_url} released!")
79-
80-
async def _process_single_list(self, mailing_list: MailingList):
81-
logger.info("Processing mailing list: {}", mailing_list.source_url)
82-
state = ListState.FAILED
83-
84-
try:
85-
list_dir = await ensure_mirror(mailing_list.name, mailing_list.source_url)
86-
heads = dict(mailing_list.last_processed_heads)
87-
activities_db = []
88-
activities_kafka = []
89-
90-
for shard_path in discover_shards(list_dir):
91-
shard = shard_index(shard_path)
92-
commit_ids = await new_commits(shard_path, heads.get(shard))
93-
for git_id in commit_ids:
94-
heads[shard] = git_id
95-
try:
96-
message, blob_id = await asyncio.to_thread(read_email, shard_path, git_id)
97-
parsed = parse_email(
98-
message,
99-
mailing_list.source_url,
100-
mailing_list.name,
101-
git_id,
102-
blob_id,
103-
mailing_list.segment_id,
104-
mailing_list.integration_id,
105-
)
106-
except Exception as e:
107-
logger.error(
108-
"Skipping unparseable commit {} in shard {}: {}",
109-
git_id,
110-
shard_path,
111-
repr(e),
112-
)
113-
continue
114-
activity_data = parsed["activityData"]
115-
if not activity_data["timestamp"]:
116-
logger.warning(
117-
"Skipping message {} with no parseable Date header",
118-
activity_data["sourceId"],
119-
)
120-
continue
121-
activity_data["segmentId"] = mailing_list.segment_id
122-
123-
# Deterministic (not random) so a retried batch after a crash or a
124-
# Kafka-send failure re-inserts the exact same row instead of a
125-
# duplicate, per commit id which is stable across retries.
126-
result_id = str(
127-
uuid.uuid5(
128-
uuid.NAMESPACE_URL,
129-
f"{mailing_list.integration_id}:{shard}:{git_id}",
130-
)
100+
logger.error(
101+
"Skipping unparseable commit {} in shard {}: {}",
102+
git_id,
103+
shard_path,
104+
repr(e),
105+
)
106+
continue
107+
activity_data = parsed["activityData"]
108+
if not activity_data["timestamp"]:
109+
logger.warning(
110+
"Skipping message {} with no parseable Date header",
111+
activity_data["sourceId"],
131112
)
132-
data_dict = {"type": IntegrationResultType.ACTIVITY, "data": activity_data}
133-
activities_db.append(
134-
(
135-
result_id,
136-
IntegrationResultState.PENDING,
137-
data_dict,
138-
DEFAULT_TENANT_ID,
139-
mailing_list.integration_id,
140-
)
113+
continue
114+
activity_data["segmentId"] = mailing_list.segment_id
115+
116+
# Deterministic (not random) so a retried batch after a crash or a
117+
# Kafka-send failure re-inserts the exact same row instead of a
118+
# duplicate, per commit id which is stable across retries.
119+
result_id = str(
120+
uuid.uuid5(
121+
uuid.NAMESPACE_URL,
122+
f"{mailing_list.integration_id}:{shard}:{git_id}",
141123
)
142-
activities_kafka.append(
143-
queue_service.build_activity_kafka_message(
144-
mailing_list.segment_id, mailing_list.integration_id, result_id
145-
)
124+
)
125+
data_dict = {"type": IntegrationResultType.ACTIVITY, "data": activity_data}
126+
activities_db.append(
127+
(
128+
result_id,
129+
IntegrationResultState.PENDING,
130+
data_dict,
131+
DEFAULT_TENANT_ID,
132+
mailing_list.integration_id,
146133
)
134+
)
135+
activities_kafka.append(
136+
queue_service.build_activity_kafka_message(
137+
mailing_list.segment_id, mailing_list.integration_id, result_id
138+
)
139+
)
140+
141+
if len(activities_db) >= ACTIVITY_FLUSH_BATCH_SIZE:
142+
await batch_insert_activities(activities_db)
143+
await queue_service.send_batch_activities(activities_kafka)
144+
await update_processed_heads(mailing_list.id, heads)
145+
activities_db = []
146+
activities_kafka = []
147+
148+
if activities_db:
149+
await batch_insert_activities(activities_db)
150+
await queue_service.send_batch_activities(activities_kafka)
151+
152+
await update_processed_heads(mailing_list.id, heads)
153+
state = ListState.COMPLETED
154+
except Exception as e:
155+
logger.error(f"Processing failed for list {mailing_list.source_url}: {repr(e)}")
156+
state = ListState.FAILED
157+
finally:
158+
logger.info(f"Updating list {mailing_list.source_url} state to {state}")
159+
await mark_list_processed(mailing_list.id, state)
147160

148-
if len(activities_db) >= ACTIVITY_FLUSH_BATCH_SIZE:
149-
await batch_insert_activities(activities_db)
150-
await queue_service.send_batch_activities(activities_kafka)
151-
await update_processed_heads(mailing_list.id, heads)
152-
activities_db = []
153-
activities_kafka = []
154-
155-
if activities_db:
156-
await batch_insert_activities(activities_db)
157-
await queue_service.send_batch_activities(activities_kafka)
158-
159-
await update_processed_heads(mailing_list.id, heads)
160-
state = ListState.COMPLETED
161-
except Exception as e:
162-
logger.error(f"Processing failed for list {mailing_list.source_url}: {repr(e)}")
163-
state = ListState.FAILED
164-
finally:
165-
logger.info(f"Updating list {mailing_list.source_url} state to {state}")
166-
await mark_list_processed(mailing_list.id, state)
167-
168-
logger.info("Completed processing mailing list: {}", mailing_list.source_url)
161+
logger.info("Completed processing mailing list: {}", mailing_list.source_url)

0 commit comments

Comments
 (0)