diff --git a/docsgpt/worker.py b/docsgpt/worker.py index 5e70393c..0e5242bb 100755 --- a/docsgpt/worker.py +++ b/docsgpt/worker.py @@ -1519,7 +1519,7 @@ def sync_worker(self, frequency): with db_readonly() as conn: result = conn.execute( sql_text( - "SELECT id, name, user_id, type, remote_data, retriever, connection_id " + "SELECT id, name, user_id, type, remote_data, retriever, connection_id, metadata " "FROM sources WHERE sync_frequency = :freq" ), {"freq": frequency}, @@ -1549,6 +1549,23 @@ def sync_worker(self, frequency): sync_counts["sync_skipped"] += 1 continue + metadata = doc.get("metadata") + if isinstance(metadata, str): + try: + metadata = json.loads(metadata) + except ValueError: + metadata = {} + if ( + doc.get("connection_id") + and isinstance(metadata, dict) + and metadata.get("sync_state") == "paused_reconnect" + ): + # An S3 or GitHub source whose connection needs reconnecting: + # it resumes when the owner reconnects, rather than failing (and + # notifying) on every schedule until then. + sync_counts["sync_skipped"] += 1 + continue + source_data = normalize_remote_data(source_type, doc.get("remote_data")) if not source_data: # No syncable URL/config — skip instead of dispatching a sync diff --git a/tests/connectors/test_runtime.py b/tests/connectors/test_runtime.py index cde70b67..3c53e415 100644 --- a/tests/connectors/test_runtime.py +++ b/tests/connectors/test_runtime.py @@ -239,6 +239,27 @@ class TestScheduledSync: assert counts["sync_dispatched"] == 1 assert counts["sync_skipped"] == 1 + def test_paused_repository_is_skipped_until_reconnected(self, pg_conn): + """A GitHub or S3 source paused for reconnect is not retried (and failed) on every schedule.""" + from docsgpt import worker + + cid = _connection(pg_conn, provider="github", status="reconnect_needed", + secrets={"credentials": {"access_token": "revoked"}}) + pg_conn.execute(text( + "INSERT INTO sources (user_id, name, type, sync_frequency, connection_id, remote_data, metadata) " + "VALUES ('alice', 'acme/api', 'github', 'daily', CAST(:c AS uuid), '{\"repo_url\": \"acme/api\"}', " + "'{\"sync_state\": \"paused_reconnect\"}')" + ), {"c": cid}) + + @contextmanager + def _yield(): + yield pg_conn + + with patch.object(worker, "db_readonly", _yield), patch.object(worker, "sync") as sync: + counts = worker.sync_worker(MagicMock(), "daily") + sync.assert_not_called() + assert counts["sync_skipped"] == 1 + def test_paused_connection_is_not_synced(self, pg_conn): from docsgpt import worker