code: async def _run(self):
          
  
          tick_gen = _task_idle_ticks(0.5*self._shard_ttl)
          self._is_running = True
          self._ready_event.clear()
          
          while True:
              try:
                  await self._lease_shard()
                  break
              except grpc.RpcError as exc:
                  nap = _rand_uniform(3, 15)
                  logger.warn(f'failed in gRPC [{exc.code()}]: {exc.details()} '
                              f'. napping {nap:.0f} secs ...')
  
                  if await self._continueAfterSleep(nap):
                      continue
                  else:
                      return
  
          assert self._shard_id is not None           
                 
          try:
              while self._is_running:
  
                  self._ready_event.clear()
                  try:
                      await self._renew_timestamp()
                      await self._keepalive_shard()
                  except grpc.RpcError as exc: 
                      # exc.code()==grpc.StatusCode.UNAVAILABLE
                      nap = _rand_uniform(3, 15)
                      logger.warn(f'failed in grpc[{exc.code()}]: {exc.details()}'
                                  f', napping {nap:.0f}secs ...')
  
                      if await self._continueAfterSleep(nap):
                          continue
                      else:
                          break
  
                  self._ready_event.set()
  
                  if await self._continueAfterSleep(next(tick_gen)):
                      continue
                  else:
                      break
  
          except asyncio.CancelledError:
              pass
  
          except Exception:
              logger.error(f'Error in shard#{self._shard_id}:', exc_info=True)
  
          finally:
              self._ready_event.clear()
              await self._lease.revoke() # 取消租约
              logger.debug(f'shard#{self._shard_id}, the lease revoked')
docs: 后台任务更新时间戳和重置序号