opal_server/server.py
async def start_server_background_tasks(self):
        """
        starts the background processes (as asyncio tasks) if such are configured.

        all workers will start these tasks:
        - publisher: a client that is used to publish updates to the client.

        only the leader worker (first to obtain leadership lock) will start these tasks:
        - (repo) watcher: monitors the policy git repository for changes.
        """
        if self.publisher is not None:
            async with self.publisher:
                if self._init_policy_watcher:
                    # repo watcher is enabled, but we want only one worker to run it
                    # (otherwise for each new commit, we will publish multiple updates via pub/sub).
                    # leadership is determined by the first worker to obtain a lock
                    self.leadership_lock = NamedLock(opal_server_config.LEADER_LOCK_FILE_PATH)
                    async with self.leadership_lock:
                        # only one worker gets here, the others block. in case the leader worker
                        # is terminated, another one will obtain the lock and become leader.
                        logger.info("leadership lock acquired, leader pid: {pid}", pid=os.getpid())
                        logger.info("listening on webhook topic: '{topic}'",
                                    topic=opal_server_config.POLICY_REPO_WEBHOOK_TOPIC)
                        # init policy watcher
                        if self.watcher is None:
                            # only the leader should discard previous clones
                            self.create_local_clone_path_and_discard_previous_clones()
                            self.watcher = setup_watcher_task(self.publisher, remote_source_url=self._policy_remote_url)
                        # the leader listens to the webhook topic (webhook api route can be hit randomly in all workers)
                        # and triggers the watcher to check for changes in the tracked upstream remote.
                        await self.pubsub.endpoint.subscribe([opal_server_config.POLICY_REPO_WEBHOOK_TOPIC], partial(trigger_repo_watcher_pull, self.watcher))
                        # running the watcher, and waiting until it stops (until self.watcher.signal_stop() is called)
                        async with self.watcher:
                            if self.broadcast_keepalive is not None:
                                self.broadcast_keepalive.start()
                            await self.watcher.wait_until_should_stop()

        starts the background processes (as asyncio tasks) if such are configured.

        all workers will start these tasks:
        - publisher: a client that is used to publish updates to the client.

        only the leader worker (first to obtain leadership lock) will start these tasks:
        - (repo) watcher: monitors the policy git repository for changes.
        
starts the background processes (as asyncio tasks) if such are configured.
all workers will start these tasks:
publisher: a client that is used to publish updates to the client.

only the leader worker (first to obtain leadership lock) will start these tasks:
(repo) watcher: monitors the policy git repository for changes.
starts the background processes ( as asyncio tasks ) if such are configured . all workers will start these tasks : publisher : a client that is used to publish updates to the client . only the leader worker ( first to obtain leadership lock ) will start these tasks : ( repo ) watcher : monitors the policy git repository for changes .
async def start_server_background_tasks(self):
        if self.publisher is not None:
            async with self.publisher:
                if self._init_policy_watcher:
                    self.leadership_lock = NamedLock(opal_server_config.LEADER_LOCK_FILE_PATH)
                    async with self.leadership_lock:
                        logger.info("leadership lock acquired, leader pid: {pid}", pid=os.getpid())
                        logger.info("listening on webhook topic: '{topic}'",
                                    topic=opal_server_config.POLICY_REPO_WEBHOOK_TOPIC)
                        if self.watcher is None:
                            self.create_local_clone_path_and_discard_previous_clones()
                            self.watcher = setup_watcher_task(self.publisher, remote_source_url=self._policy_remote_url)
                        await self.pubsub.endpoint.subscribe([opal_server_config.POLICY_REPO_WEBHOOK_TOPIC], partial(trigger_repo_watcher_pull, self.watcher))
                        async with self.watcher:
                            if self.broadcast_keepalive is not None:
                                self.broadcast_keepalive.start()
                            await self.watcher.wait_until_should_stop()
async def start_server_background_tasks ( self ) : if self . publisher is not None : async with self . publisher : if self . _init_policy_watcher : self . leadership_lock = NamedLock ( opal_server_config . LEADER_LOCK_FILE_PATH ) async with self . leadership_lock : logger . info ( "leadership lock acquired, leader pid: {pid}" , pid = os . getpid ( ) ) logger . info ( "listening on webhook topic: '{topic}'" , topic = opal_server_config . POLICY_REPO_WEBHOOK_TOPIC ) if self . watcher is None : self . create_local_clone_path_and_discard_previous_clones ( ) self . watcher = setup_watcher_task ( self . publisher , remote_source_url = self . _policy_remote_url ) await self . pubsub . endpoint . subscribe ( [ opal_server_config . POLICY_REPO_WEBHOOK_TOPIC ] , partial ( trigger_repo_watcher_pull , self . watcher ) ) async with self . watcher : if self . broadcast_keepalive is not None : self . broadcast_keepalive . start ( ) await self . watcher . wait_until_should_stop ( )
starts the background processes (as asyncio tasks) if such are configured.
starts the background processes ( as asyncio tasks ) if such are configured .
"""
        starts the background processes (as asyncio tasks) if such are configured.

        all workers will start these tasks:
        - publisher: a client that is used to publish updates to the client.

        only the leader worker (first to obtain leadership lock) will start these tasks:
        - (repo) watcher: monitors the policy git repository for changes.
        """ # repo watcher is enabled, but we want only one worker to run it # (otherwise for each new commit, we will publish multiple updates via pub/sub). # leadership is determined by the first worker to obtain a lock # only one worker gets here, the others block. in case the leader worker # is terminated, another one will obtain the lock and become leader. # init policy watcher # only the leader should discard previous clones # the leader listens to the webhook topic (webhook api route can be hit randomly in all workers) # and triggers the watcher to check for changes in the tracked upstream remote. # running the watcher, and waiting until it stops (until self.watcher.signal_stop() is called)
self