I am using Redis OM (redis-om==0.3.5) for my chess like game where two people join and make moves one by one. The game is CacheGame and player is CacheGamePlayer. The issue is on high concurrency, lets say from 30000 games per day I am seeing following discrepancies for like 90-120 times. Game ID XMHFS-NGKd: Turn Discrepancy detected - move_count=18 move_history_length=18. Moves: ['b3e3 599/120 600/120 1', 'b10c8 599/120 595/120 5', 'b1c3 597/120 595/120 2', 'h10g8 597/ .... What that means is some times we are some how are getting staled data, likely from following places latest_game = CacheGame.get(game_id) self.game = CacheGame.get(game_id) Here’s a rephrased version of your text with smoother flow and clarity: We’re running Flask on the backend with eventlet monkey patching, which relies on greenlets. The purpose of using pipeline.watch and multi/execute is to avoid working with stale data. if another process updates the key, it should trigger a WatchError. My concern is that since I’m creating a separate connection to Redis OM, this might be why the data isn’t always up to date. The goal is to ensure that what I store in Redis OM is immediately reflected in memory. Could you verify if this is indeed the root cause, or if something else might be responsible? @function_trace(name='apply_move') def apply_move(self): """ Apply a move using a Check-and-Set (CAS) pattern to ensure atomicity. This function will WATCH the game's Redis key. If the key is modified by another process before this transaction completes, a WatchError is raised, and the operation is retried with the fresh game data. """ client = CacheGame.Meta.database game_id = self.game.game_id for attempt in range(MAX_CAS_RETRIES): try: # Use a pipeline to group WATCH, GET, and the transaction with client.pipeline() as pipe: game_key = CacheGame.make_primary_key(game_id) pipe.watch(game_key) latest_game = CacheGame.get(game_id) self.game = latest_game self.board.fen = latest_game.fen self.validate_move() self.set_move_result() self.apply_move_updates() # This will fail and raise WatchError if `game_key` was changed. pipe.multi() self.game.save(pipeline=pipe) result = pipe.execute() # Ensure exactly one command succeeded if not result or len(result) != 1 or result[0] is not True: raise PersistenceError # Persisted state is now our current state. self.game = CacheGame.get(game_id) self.execute_post_move_operations() if self.game.state == GameStates.END.value: celery.send_task(GameTasks.GAME_END_TASK.value, args=(game_id,)) return except WatchError: logger.error( f'CATCH AND SET conflict on game {game_id} attempt {attempt + 1}/{MAX_CAS_RETRIES}. Retrying...' ) time.sleep(CAS_RETRY_DELAY_SECS) continue except (RedisError, PersistenceError) as err: logger.error(f'PersistenceError during move on game {game_id}: {err}', exc_info=True) self.emit_error_response() return def apply_move_updates(self): current_time = get_current_time_in_ms() self.elapsed_secs = self.game.get_elapsed_secs(current_time, self.thinking_seconds) if self.game.is_timed: self.game.update_player_timers(self.elapsed_secs) self.game.fen = self.board.fen self.game.server_clock = current_time self.game.move_count += 1 self.game.last_move_timestamp = datetime.now() # Expiry time must be updated once fen and server_clock set self.game.expiry_time = round(time.time() + self.game.get_game_expiry_secs() + GAME_EXPIRY_GRACE_SECS) if not self.game.bot_game: self.set_requester_states() self.formatted_move = self.get_formatted_move(self.elapsed_secs) self.pre_move = self.game.pre_move self.game.pre_move = None self.set_move_history() if self.game.move_count > len(self.game.move_history): logger.error('Discrepancy detected: move count exceeds recorded move history length.') logger.error( f'Game ID {self.game.game_id}: Discrepancy detected - move_count={self.game.move_count} exceeds ' f'move_history_length={len(self.game.move_history)}. Moves: {self.game.move_history}. ' f'Snapshot: server_clock={self.game.server_clock}, expiry_time={self.game.expiry_time}, ' f'current_turn={self.game.turn_now}, fen={self.game.fen}') end_reason = self.move_result.get('end_reason') if end_reason: game_result = ( f'{self.move_side.label}_{GameResult.WON.label}' if end_reason in MATE_REASONS else GameResult.DRAW.label ) self.game.end_game(end_reason, game_result=game_result, save=False) Continue reading...
