starred/onyx
1
0 Fork 0
mirror of https://github.com/onyx-dot-app/onyx.git synced 2026-10-19 01:50:51 -05:00
Code Issues Packages Projects Releases 10 Wiki Activity

chore(permissions): remove the Redis state for permission and group syncs (#15977)

Browse Source 
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Evan LohnandClaude Opus 5.5 authored and GitHub committed 2026-10-16 03:25:36 +00:00
1 parent e3f1fb21ea
commit 308ad098e4
13 files changed
+9 -583
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.agents/feature-map/components/background-jobs.md
-1
View File
@@ -217,7 +217,6 @@ seconds, `beat_schedule.py:BEAT_EXPIRES_DEFAULT`), matching the
Every connector-scoped key is namespaced per tenant through the standard Redis
client factory and, within a tenant, per `cc_pair_id` via the `RedisConnector`
family in `onyx/redis/` (`redis_connector_prune.py`, `redis_connector_delete.py`,
`redis_connector_doc_perm_sync.py`, `redis_connector_ext_group_sync.py`,
`redis_document_set.py`, `redis_usergroup.py`).
| Pattern | Example owner | TTL | Meaning |
backend/onyx/background/celery/apps/app_base.py
-18
View File
@@ -39,8 +39,6 @@ from onyx.configs.constants import ONYX_CLOUD_CELERY_TASK_PREFIX, OnyxRedisLocks
from onyx.db.engine.sql_engine import get_sqlalchemy_engine
from onyx.redis.redis_connector import RedisConnector
from onyx.redis.redis_connector_delete import RedisConnectorDelete
from onyx.redis.redis_connector_doc_perm_sync import RedisConnectorPermissionSync
from onyx.redis.redis_connector_ext_group_sync import RedisConnectorExternalGroupSync
from onyx.redis.redis_connector_prune import RedisConnectorPrune
from onyx.redis.redis_document_set import RedisDocumentSet
from onyx.redis.redis_pool import get_redis_client
@@ -240,22 +238,6 @@ def on_task_postrun(
RedisConnectorPrune.remove_from_taskset(int(cc_pair_id), task_id, r)
return
if task_id.startswith(RedisConnectorPermissionSync.SUBTASK_PREFIX):
cc_pair_id = RedisConnector.get_id_from_task_id(task_id)
if cc_pair_id is not None:
RedisConnectorPermissionSync.remove_from_taskset(
int(cc_pair_id), task_id, r
)
return
if task_id.startswith(RedisConnectorExternalGroupSync.SUBTASK_PREFIX):
cc_pair_id = RedisConnector.get_id_from_task_id(task_id)
if cc_pair_id is not None:
RedisConnectorExternalGroupSync.remove_from_taskset(
int(cc_pair_id), task_id, r
)
return
def on_task_revoked(
request: Any | None = None,
backend/onyx/background/celery/apps/primary.py
-4
View File
@@ -29,8 +29,6 @@ from onyx.db.engine.sql_engine import SqlEngine, get_session_with_current_tenant
from onyx.db.index_attempt import get_index_attempt, mark_attempt_canceled
from onyx.db.indexing_coordination import IndexingCoordination
from onyx.redis.redis_connector_delete import RedisConnectorDelete
from onyx.redis.redis_connector_doc_perm_sync import RedisConnectorPermissionSync
from onyx.redis.redis_connector_ext_group_sync import RedisConnectorExternalGroupSync
from onyx.redis.redis_connector_prune import RedisConnectorPrune
from onyx.redis.redis_connector_stop import RedisConnectorStop
from onyx.redis.redis_document_set import RedisDocumentSet
@@ -197,8 +195,6 @@ def on_worker_init(sender: Worker, **kwargs: Any) -> None:
RedisConnectorDelete.reset_all(r)
RedisConnectorPrune.reset_all(r)
RedisConnectorStop.reset_all(r)
RedisConnectorPermissionSync.reset_all(r)
RedisConnectorExternalGroupSync.reset_all(r)
# mark orphaned index attempts as failed
# This uses database coordination instead of Redis fencing
backend/onyx/configs/constants.py
-28
View File
@@ -216,24 +216,6 @@ CELERY_TASK_WAIT_FOR_FENCE_TIMEOUT = 5 * 60 # 5 min
# if we can get callbacks as object bytes download, we could lower this a lot.
CELERY_PRUNING_LOCK_TIMEOUT = 3600 # 1 hour (in seconds)
# Held for one connector's whole document permission sync and refreshed every
# quarter of the generic beat TTL, so it must not be set below that TTL. A
# connector silent longer than this loses it, so large tenants raise it.
CELERY_PERMISSIONS_SYNC_LOCK_TIMEOUT: int = lock_timeout_from_env(
"CELERY_PERMISSIONS_SYNC_LOCK_TIMEOUT",
3600,
minimum=CELERY_GENERIC_BEAT_LOCK_TIMEOUT,
)
# While this lock is held, duplicate dispatches for the same cc_pair exit
# immediately. Deployments whose group syncs legitimately run for hours should
# raise this toward the JOB_TIMEOUT crawl deadline (6h) so re-dispatches can't
# stack concurrent crawls on one heavy worker; a worker that dies mid-sync
# leaves the lock stuck for at most this TTL.
CELERY_EXTERNAL_GROUP_SYNC_LOCK_TIMEOUT: int = lock_timeout_from_env(
"CELERY_EXTERNAL_GROUP_SYNC_LOCK_TIMEOUT", 300
)
CELERY_USER_FILE_PROCESSING_LOCK_TIMEOUT = 30 * 60 # 30 minutes (in seconds)
# How long a queued user-file task is valid before workers discard it.
@@ -572,10 +554,6 @@ class OnyxRedisLocks:
CHECK_AVAILABLE_TENANTS_LOCK = "da_lock:check_available_tenants"
CLOUD_PRE_PROVISION_TENANT_LOCK = "da_lock:pre_provision_tenant"
CONNECTOR_DOC_PERMISSIONS_SYNC_LOCK_PREFIX = (
"da_lock:connector_doc_permissions_sync"
)
CONNECTOR_EXTERNAL_GROUP_SYNC_LOCK_PREFIX = "da_lock:connector_external_group_sync"
PRUNING_LOCK_PREFIX = "da_lock:pruning"
ORPHAN_TAG_SWEEP_LOCK = "da_lock:orphan_tag_sweep"
INDEXING_METADATA_PREFIX = "da_metadata:indexing"
@@ -616,12 +594,6 @@ class OnyxRedisLocks:
class OnyxRedisSignals:
BLOCK_VALIDATE_INDEXING_FENCES = "signal:block_validate_indexing_fences"
BLOCK_VALIDATE_EXTERNAL_GROUP_SYNC_FENCES = (
"signal:block_validate_external_group_sync_fences"
)
BLOCK_VALIDATE_PERMISSION_SYNC_FENCES = (
"signal:block_validate_permission_sync_fences"
)
BLOCK_PRUNING = "signal:block_pruning"
# Set by a sweep that skipped because another one held the lock.
ORPHAN_TAG_SWEEP_PENDING = "signal:orphan_tag_sweep_pending"
backend/onyx/redis/redis_connector.py
-8
View File
@@ -1,6 +1,4 @@
from onyx.redis.redis_connector_delete import RedisConnectorDelete
from onyx.redis.redis_connector_doc_perm_sync import RedisConnectorPermissionSync
from onyx.redis.redis_connector_ext_group_sync import RedisConnectorExternalGroupSync
from onyx.redis.redis_connector_prune import RedisConnectorPrune
from onyx.redis.redis_connector_stop import RedisConnectorStop
from onyx.redis.redis_pool import get_redis_client
@@ -22,12 +20,6 @@ class RedisConnector:
self.stop = RedisConnectorStop(tenant_id, cc_pair_id, self.redis)
self.prune = RedisConnectorPrune(tenant_id, cc_pair_id, self.redis)
self.delete = RedisConnectorDelete(tenant_id, cc_pair_id, self.redis)
self.permissions = RedisConnectorPermissionSync(
tenant_id, cc_pair_id, self.redis
)
self.external_group_sync = RedisConnectorExternalGroupSync(
tenant_id, cc_pair_id, self.redis
)
@staticmethod
def get_id_from_fence_key(key: str) -> str | None:
backend/onyx/redis/redis_connector_doc_perm_sync.py
-300
View File
@@ -1,300 +0,0 @@
import time
from datetime import datetime
from logging import Logger
from typing import Any, NamedTuple, cast
from pydantic import BaseModel
from redis.lock import Lock as RedisLock
from onyx.access.models import DocExternalAccess, ElementExternalAccess
from onyx.configs.constants import (
CELERY_GENERIC_BEAT_LOCK_TIMEOUT,
CELERY_PERMISSIONS_SYNC_LOCK_TIMEOUT,
OnyxRedisConstants,
)
from onyx.redis.redis_pool import SCAN_ITER_COUNT_DEFAULT
from onyx.redis.tenant_redis_client import TenantRedisClient
from onyx.server.metrics.perm_sync_metrics import (
observe_doc_perm_sync_db_update_duration,
)
from onyx.utils.variable_functionality import fetch_versioned_implementation
class PermissionSyncResult(NamedTuple):
"""Result of a permission sync operation.
Attributes:
num_updated: Number of documents successfully updated
num_errors: Number of documents that failed to update
"""
num_updated: int
num_errors: int
class RedisConnectorPermissionSyncPayload(BaseModel):
id: str
submitted: datetime
started: datetime | None
celery_task_id: str | None
class RedisConnectorPermissionSync:
"""Manages interactions with redis for doc permission sync tasks. Should only be accessed
through RedisConnector."""
PREFIX = "connectordocpermissionsync"
FENCE_PREFIX = f"{PREFIX}_fence"
FENCE_TTL = 7 * 24 * 60 * 60 # 7 days - defensive TTL to prevent memory leaks
# phase 1 - geneartor task and progress signals
GENERATORTASK_PREFIX = f"{PREFIX}+generator" # connectorpermissions+generator
GENERATOR_PROGRESS_PREFIX = (
PREFIX + "_generator_progress"
) # connectorpermissions_generator_progress
GENERATOR_COMPLETE_PREFIX = (
PREFIX + "_generator_complete"
) # connectorpermissions_generator_complete
TASKSET_PREFIX = f"{PREFIX}_taskset" # connectorpermissions_taskset
SUBTASK_PREFIX = f"{PREFIX}+sub" # connectorpermissions+sub
# used to signal the overall workflow is still active
# it's impossible to get the exact state of the system at a single point in time
# so we need a signal with a TTL to bridge gaps in our checks
ACTIVE_PREFIX = PREFIX + "_active"
ACTIVE_TTL = CELERY_PERMISSIONS_SYNC_LOCK_TIMEOUT * 2
def __init__(self, tenant_id: str, id: int, redis: TenantRedisClient) -> None:
self.tenant_id: str = tenant_id
self.id = id
self.redis = redis
self.fence_key: str = f"{self.FENCE_PREFIX}_{id}"
self.generator_task_key = f"{self.GENERATORTASK_PREFIX}_{id}"
self.generator_progress_key = f"{self.GENERATOR_PROGRESS_PREFIX}_{id}"
self.generator_complete_key = f"{self.GENERATOR_COMPLETE_PREFIX}_{id}"
self.taskset_key = f"{self.TASKSET_PREFIX}_{id}"
self.subtask_prefix: str = f"{self.SUBTASK_PREFIX}_{id}"
self.active_key = f"{self.ACTIVE_PREFIX}_{id}"
def taskset_clear(self) -> None:
self.redis.delete(self.taskset_key)
def generator_clear(self) -> None:
self.redis.delete(self.generator_progress_key)
self.redis.delete(self.generator_complete_key)
def get_remaining(self) -> int:
return self.redis.scard(self.taskset_key)
def get_active_task_count(self) -> int:
"""Count of active permission sync tasks"""
count = 0
for _ in self.redis.sscan_iter(
OnyxRedisConstants.ACTIVE_FENCES,
RedisConnectorPermissionSync.FENCE_PREFIX + "*",
count=SCAN_ITER_COUNT_DEFAULT,
):
count += 1
return count
@property
def fenced(self) -> bool:
return bool(self.redis.exists(self.fence_key))
@property
def payload(self) -> RedisConnectorPermissionSyncPayload | None:
# read related data and evaluate/print task progress
fence_bytes = cast(Any, self.redis.get(self.fence_key))
if fence_bytes is None:
return None
fence_str = fence_bytes.decode("utf-8")
payload = RedisConnectorPermissionSyncPayload.model_validate_json(
cast(str, fence_str)
)
return payload
def set_fence(
self,
payload: RedisConnectorPermissionSyncPayload | None,
) -> None:
if not payload:
self.redis.srem(OnyxRedisConstants.ACTIVE_FENCES, self.fence_key)
self.redis.delete(self.fence_key)
return
self.redis.set(self.fence_key, payload.model_dump_json(), ex=self.FENCE_TTL)
self.redis.sadd(OnyxRedisConstants.ACTIVE_FENCES, self.fence_key)
def set_active(self) -> None:
"""This sets a signal to keep the permissioning flow from getting cleaned up within
the expiration time.
The slack in timing is needed to avoid race conditions where simply checking
the celery queue and task status could result in race conditions."""
self.redis.set(self.active_key, 0, ex=self.ACTIVE_TTL)
def active(self) -> bool:
return bool(self.redis.exists(self.active_key))
@property
def generator_complete(self) -> int | None:
"""the fence payload is an int representing the starting number of
permission sync tasks to be processed ... just after the generator completes."""
fence_bytes = self.redis.get(self.generator_complete_key)
if fence_bytes is None:
return None
if fence_bytes == b"None":
return None
fence_int = int(cast(bytes, fence_bytes).decode())
return fence_int
@generator_complete.setter
def generator_complete(self, payload: int | None) -> None:
"""Set the payload to an int to set the fence, otherwise if None it will
be deleted"""
if payload is None:
self.redis.delete(self.generator_complete_key)
return
self.redis.set(self.generator_complete_key, payload, ex=self.FENCE_TTL)
def update_db(
self,
lock: RedisLock | None,
new_permissions: list[ElementExternalAccess],
source_string: str,
connector_id: int,
credential_id: int,
task_logger: Logger | None = None,
) -> PermissionSyncResult:
"""Update permissions for documents and hierarchy nodes.
Returns:
PermissionSyncResult containing counts of successful updates and errors
"""
last_lock_time = time.monotonic()
element_update_permissions_fn = fetch_versioned_implementation(
"onyx.background.celery.tasks.doc_permission_syncing.tasks",
"element_update_permissions",
)
num_permissions = 0
num_errors = 0
cumulative_db_update_time = 0.0
for permissions in new_permissions:
current_time = time.monotonic()
if lock and current_time - last_lock_time >= (
CELERY_GENERIC_BEAT_LOCK_TIMEOUT / 4
):
lock.reacquire()
last_lock_time = current_time
if (
permissions.external_access.num_entries
> permissions.external_access.MAX_NUM_ENTRIES
):
num_errors += 1
if task_logger:
num_users = len(permissions.external_access.external_user_emails)
num_groups = len(
permissions.external_access.external_user_group_ids
)
element_id = (
permissions.doc_id
if isinstance(permissions, DocExternalAccess)
else permissions.raw_node_id
)
task_logger.warning(
"Permissions length exceeded, skipping...: "
"%s "
"num_users=%s num_groups=%s "
"permissions.external_access.MAX_NUM_ENTRIES=%s",
element_id,
num_users,
num_groups,
permissions.external_access.MAX_NUM_ENTRIES,
)
continue
# NOTE(rkuo): this used to fire a task instead of directly writing to the DB,
# but the permissions can be excessively large if sent over the wire.
# On the other hand, the downside of doing db updates here is that we can
# block and fail if we can't make the calls to the DB ... but that's probably
# a rare enough case to be acceptable.
# This can internally exception due to db issues but still continue
# Catch exceptions per-element to avoid breaking the entire sync
db_start = time.monotonic()
try:
element_update_permissions_fn(
self.tenant_id,
permissions,
source_string,
connector_id,
credential_id,
)
num_permissions += 1
except Exception:
num_errors += 1
if task_logger:
element_id = (
permissions.doc_id
if isinstance(permissions, DocExternalAccess)
else permissions.raw_node_id
)
task_logger.exception(
"Failed to update permissions for element %s", element_id
)
finally:
cumulative_db_update_time += time.monotonic() - db_start
observe_doc_perm_sync_db_update_duration(
cumulative_db_update_time, source_string
)
return PermissionSyncResult(num_updated=num_permissions, num_errors=num_errors)
def reset(self) -> None:
self.redis.srem(OnyxRedisConstants.ACTIVE_FENCES, self.fence_key)
self.redis.delete(self.active_key)
self.redis.delete(self.generator_progress_key)
self.redis.delete(self.generator_complete_key)
self.redis.delete(self.taskset_key)
self.redis.delete(self.fence_key)
@staticmethod
def remove_from_taskset(id: int, task_id: str, r: TenantRedisClient) -> None:
taskset_key = f"{RedisConnectorPermissionSync.TASKSET_PREFIX}_{id}"
r.srem(taskset_key, task_id)
return
@staticmethod
def reset_all(r: TenantRedisClient) -> None:
"""Deletes all redis values for all connectors"""
for key in r.scan_iter(RedisConnectorPermissionSync.ACTIVE_PREFIX + "*"):
r.delete(key)
for key in r.scan_iter(RedisConnectorPermissionSync.TASKSET_PREFIX + "*"):
r.delete(key)
for key in r.scan_iter(
RedisConnectorPermissionSync.GENERATOR_COMPLETE_PREFIX + "*"
):
r.delete(key)
for key in r.scan_iter(
RedisConnectorPermissionSync.GENERATOR_PROGRESS_PREFIX + "*"
):
r.delete(key)
for key in r.scan_iter(RedisConnectorPermissionSync.FENCE_PREFIX + "*"):
r.delete(key)
backend/onyx/redis/redis_connector_ext_group_sync.py
-191
View File
@@ -1,191 +0,0 @@
from datetime import datetime
from typing import cast
from celery import Celery
from pydantic import BaseModel
from redis.lock import Lock as RedisLock
from sqlalchemy.orm import Session
from onyx.configs.constants import OnyxRedisConstants
from onyx.redis.redis_pool import SCAN_ITER_COUNT_DEFAULT
from onyx.redis.tenant_redis_client import TenantRedisClient
class RedisConnectorExternalGroupSyncPayload(BaseModel):
id: str
submitted: datetime
started: datetime | None
celery_task_id: str | None
class RedisConnectorExternalGroupSync:
"""Manages interactions with redis for external group syncing tasks. Should only be accessed
through RedisConnector."""
PREFIX = "connectorexternalgroupsync"
FENCE_PREFIX = f"{PREFIX}_fence"
FENCE_TTL = 7 * 24 * 60 * 60 # 7 days - defensive TTL to prevent memory leaks
# phase 1 - geneartor task and progress signals
GENERATORTASK_PREFIX = f"{PREFIX}+generator" # connectorexternalgroupsync+generator
GENERATOR_PROGRESS_PREFIX = (
PREFIX + "_generator_progress"
) # connectorexternalgroupsync_generator_progress
GENERATOR_COMPLETE_PREFIX = (
PREFIX + "_generator_complete"
) # connectorexternalgroupsync_generator_complete
TASKSET_PREFIX = f"{PREFIX}_taskset" # connectorexternalgroupsync_taskset
SUBTASK_PREFIX = f"{PREFIX}+sub" # connectorexternalgroupsync+sub
# used to signal the overall workflow is still active
# it's impossible to get the exact state of the system at a single point in time
# so we need a signal with a TTL to bridge gaps in our checks
ACTIVE_PREFIX = PREFIX + "_active"
ACTIVE_TTL = 3600
def __init__(self, tenant_id: str, id: int, redis: TenantRedisClient) -> None:
self.tenant_id: str = tenant_id
self.id = id
self.redis = redis
self.fence_key: str = f"{self.FENCE_PREFIX}_{id}"
self.generator_task_key = f"{self.GENERATORTASK_PREFIX}_{id}"
self.generator_progress_key = f"{self.GENERATOR_PROGRESS_PREFIX}_{id}"
self.generator_complete_key = f"{self.GENERATOR_COMPLETE_PREFIX}_{id}"
self.taskset_key = f"{self.TASKSET_PREFIX}_{id}"
self.subtask_prefix: str = f"{self.SUBTASK_PREFIX}_{id}"
self.active_key = f"{self.ACTIVE_PREFIX}_{id}"
def taskset_clear(self) -> None:
self.redis.delete(self.taskset_key)
def generator_clear(self) -> None:
self.redis.delete(self.generator_progress_key)
self.redis.delete(self.generator_complete_key)
def get_remaining(self) -> int:
# todo: move into fence
return self.redis.scard(self.taskset_key)
def get_active_task_count(self) -> int:
"""Count of active external group syncing tasks"""
count = 0
for _ in self.redis.sscan_iter(
OnyxRedisConstants.ACTIVE_FENCES,
RedisConnectorExternalGroupSync.FENCE_PREFIX + "*",
count=SCAN_ITER_COUNT_DEFAULT,
):
count += 1
return count
@property
def fenced(self) -> bool:
return bool(self.redis.exists(self.fence_key))
@property
def payload(self) -> RedisConnectorExternalGroupSyncPayload | None:
# read related data and evaluate/print task progress
fence_bytes = self.redis.get(self.fence_key)
if fence_bytes is None:
return None
fence_str = fence_bytes.decode("utf-8")
payload = RedisConnectorExternalGroupSyncPayload.model_validate_json(fence_str)
return payload
def set_fence(
self,
payload: RedisConnectorExternalGroupSyncPayload | None,
) -> None:
if not payload:
self.redis.srem(OnyxRedisConstants.ACTIVE_FENCES, self.fence_key)
self.redis.delete(self.fence_key)
return
self.redis.set(self.fence_key, payload.model_dump_json(), ex=self.FENCE_TTL)
self.redis.sadd(OnyxRedisConstants.ACTIVE_FENCES, self.fence_key)
def set_active(self) -> None:
"""This sets a signal to keep the permissioning flow from getting cleaned up within
the expiration time.
The slack in timing is needed to avoid race conditions where simply checking
the celery queue and task status could result in race conditions."""
self.redis.set(self.active_key, 0, ex=self.ACTIVE_TTL)
def active(self) -> bool:
return bool(self.redis.exists(self.active_key))
@property
def generator_complete(self) -> int | None:
"""the fence payload is an int representing the starting number of
external group syncing tasks to be processed ... just after the generator completes.
"""
fence_bytes = self.redis.get(self.generator_complete_key)
if fence_bytes is None:
return None
if fence_bytes == b"None":
return None
fence_int = int(cast(bytes, fence_bytes).decode())
return fence_int
@generator_complete.setter
def generator_complete(self, payload: int | None) -> None:
"""Set the payload to an int to set the fence, otherwise if None it will
be deleted"""
if payload is None:
self.redis.delete(self.generator_complete_key)
return
self.redis.set(self.generator_complete_key, payload, ex=self.FENCE_TTL)
def generate_tasks(
self,
celery_app: Celery,
db_session: Session,
lock: RedisLock | None,
) -> int | None:
pass
def reset(self) -> None:
self.redis.srem(OnyxRedisConstants.ACTIVE_FENCES, self.fence_key)
self.redis.delete(self.active_key)
self.redis.delete(self.generator_progress_key)
self.redis.delete(self.generator_complete_key)
self.redis.delete(self.taskset_key)
self.redis.delete(self.fence_key)
@staticmethod
def remove_from_taskset(id: int, task_id: str, r: TenantRedisClient) -> None:
taskset_key = f"{RedisConnectorExternalGroupSync.TASKSET_PREFIX}_{id}"
r.srem(taskset_key, task_id)
return
@staticmethod
def reset_all(r: TenantRedisClient) -> None:
"""Deletes all redis values for all connectors"""
for key in r.scan_iter(RedisConnectorExternalGroupSync.ACTIVE_PREFIX + "*"):
r.delete(key)
for key in r.scan_iter(RedisConnectorExternalGroupSync.TASKSET_PREFIX + "*"):
r.delete(key)
for key in r.scan_iter(
RedisConnectorExternalGroupSync.GENERATOR_COMPLETE_PREFIX + "*"
):
r.delete(key)
for key in r.scan_iter(
RedisConnectorExternalGroupSync.GENERATOR_PROGRESS_PREFIX + "*"
):
r.delete(key)
for key in r.scan_iter(RedisConnectorExternalGroupSync.FENCE_PREFIX + "*"):
r.delete(key)
backend/onyx/redis/redis_utils.py
-3
View File
@@ -1,5 +1,4 @@
from onyx.redis.redis_connector_delete import RedisConnectorDelete
from onyx.redis.redis_connector_doc_perm_sync import RedisConnectorPermissionSync
from onyx.redis.redis_connector_prune import RedisConnectorPrune
from onyx.redis.redis_document_set import RedisDocumentSet
from onyx.redis.redis_usergroup import RedisUserGroup
@@ -15,7 +14,5 @@ def is_fence(key_bytes: bytes) -> bool:
return True
if key_str.startswith(RedisConnectorPrune.FENCE_PREFIX):
return True
if key_str.startswith(RedisConnectorPermissionSync.FENCE_PREFIX):
return True
return False
backend/scripts/debugging/onyx_redis.py
-6
View File
@@ -133,12 +133,6 @@ def onyx_redis(
redis_delete_if_exists_helper(
f"{tenant_id}:{redis_connector.prune.fence_key}", dry_run, r
)
redis_delete_if_exists_helper(
f"{tenant_id}:{redis_connector.permissions.fence_key}", dry_run, r
)
redis_delete_if_exists_helper(
f"{tenant_id}:{redis_connector.external_group_sync.fence_key}", dry_run, r
)
return 0
elif command == OnyxRedisCommand.purge_pidbox:
return purge_by_match_and_type(
cli/internal/deploy/deployfiles/embedded/docker_compose/env.template
Generated
-7
View File
@@ -272,13 +272,6 @@ MINIO_ROOT_PASSWORD=minioadmin
# CELERY_WORKER_DOCPROCESSING_CONCURRENCY=
# CELERY_WORKER_LIGHT_CONCURRENCY=
# CELERY_WORKER_LIGHT_PREFETCH_MULTIPLIER=
## Per-connector external group sync lock TTL in seconds (default 300). Raise
## toward the 6h crawl deadline if group syncs run long, so periodic
## re-dispatches can't start concurrent duplicate syncs.
# CELERY_EXTERNAL_GROUP_SYNC_LOCK_TIMEOUT=
## Per-connector document permission sync lock TTL in seconds (default 3600).
## Raise it if a sync loses its lock mid-run ("lock that's no longer owned").
# CELERY_PERMISSIONS_SYNC_LOCK_TIMEOUT=
## Time limit in seconds for one doc permission sync or group sync (default:
## JOB_TIMEOUT_SECONDS, 6h). Raise it for a source whose full sync takes longer.
# PERMISSION_SYNC_TIMEOUT_SECONDS=
deployment/docker_compose/env.template
-7
View File
@@ -272,13 +272,6 @@ MINIO_ROOT_PASSWORD=minioadmin
# CELERY_WORKER_DOCPROCESSING_CONCURRENCY=
# CELERY_WORKER_LIGHT_CONCURRENCY=
# CELERY_WORKER_LIGHT_PREFETCH_MULTIPLIER=
## Per-connector external group sync lock TTL in seconds (default 300). Raise
## toward the 6h crawl deadline if group syncs run long, so periodic
## re-dispatches can't start concurrent duplicate syncs.
# CELERY_EXTERNAL_GROUP_SYNC_LOCK_TIMEOUT=
## Per-connector document permission sync lock TTL in seconds (default 3600).
## Raise it if a sync loses its lock mid-run ("lock that's no longer owned").
# CELERY_PERMISSIONS_SYNC_LOCK_TIMEOUT=
## Time limit in seconds for one doc permission sync or group sync (default:
## JOB_TIMEOUT_SECONDS, 6h). Raise it for a source whose full sync takes longer.
# PERMISSION_SYNC_TIMEOUT_SECONDS=
deployment/helm/charts/onyx/Chart.yaml
+6 -3
View File
@@ -5,7 +5,7 @@ home: https://www.onyx.app/
sources:
- "https://github.com/onyx-dot-app/onyx"
type: application
version: 0.9.15
version: 0.9.16
appVersion: latest
annotations:
category: Productivity
@@ -21,8 +21,11 @@ annotations:
image: docker.io/onyxdotapp/onyx-backend:latest
artifacthub.io/changes: |
- kind: added
description: Add the SKIP_SPREADSHEET_CONTENT setting. When it is true, connector
indexing indexes spreadsheets and other tabular files by title only.
description: Add PERMISSION_SYNC_TIMEOUT_SECONDS, the time limit for one
document permission sync or group sync.
- kind: removed
description: Remove CELERY_PERMISSIONS_SYNC_LOCK_TIMEOUT and
CELERY_EXTERNAL_GROUP_SYNC_LOCK_TIMEOUT. These syncs now hold their lock in Postgres.
dependencies:
# Helm uses the first condition path that exists: `postgresqlOperator.enabled`
# is unset by default, so this falls through to `postgresql.enabled`. Set it
deployment/helm/charts/onyx/values.yaml
+3 -7
View File
@@ -1806,13 +1806,9 @@ configMap:
CELERY_WORKER_LIGHT_CONCURRENCY: ""
CELERY_WORKER_LIGHT_PREFETCH_MULTIPLIER: ""
CELERY_WORKER_USER_FILE_PROCESSING_CONCURRENCY: ""
# Per-connector external group sync lock TTL in seconds (default 300); raise
# toward the 6h crawl deadline if group syncs run long, so periodic
# re-dispatches can't start concurrent duplicate syncs.
CELERY_EXTERNAL_GROUP_SYNC_LOCK_TIMEOUT: ""
# Per-connector document permission sync lock TTL in seconds (default 3600);
# raise it if a sync loses its lock mid-run ("lock that's no longer owned").
CELERY_PERMISSIONS_SYNC_LOCK_TIMEOUT: ""
# Time limit in seconds for one doc permission sync or group sync (default:
# JOB_TIMEOUT_SECONDS, 6h); raise it for a source whose full sync takes longer.
PERMISSION_SYNC_TIMEOUT_SECONDS: ""
# Document-index sync beat lock TTL in seconds (default 120); raise it if the
# sync pass aborts with "no longer owned" or "Lock not owned on completion".
CELERY_DOCUMENT_SYNC_BEAT_LOCK_TIMEOUT: ""