What this PR does / why we need it:

online_write_batch performs its staleness check against a snapshot that is read
before any write is queued:

1. Read phase — hmget the stored event timestamp for every row's entity key, then
   pipe.execute() once.
2. Write phase — loop the rows, skip any row whose timestamp is <= the stored one,
   otherwise queue an hset.

Because the read phase completes before the first write is queued, rows in the same batch
that share an entity key all compare against the same pre-batch value. None of them can
see the others. When nothing is stored yet, prev_event_time is None for all of them,
the guard is skipped entirely, and every row gets an hset queued. Redis applies
pipelined commands in order, so the last row in list order wins — regardless of its
event time.

Pushing one entity key with timestamps in descending order therefore leaves the oldest
value in the store:

test_data = pd.DataFrame({
    "customer_id": [1] * 3,
    "total_purchases": [30, 20, 10],
    "event_timestamp": [t3, t2, t1],   # latest first
})
store.push_features(fg_name, test_data, mode="online")
# get_online_features returns 10, the value belonging to t1

No error and no warning — just a silently incorrect feature value.

The fix

Track the latest event timestamp queued so far per entity key within the batch, and
compare each row against whichever is newer: the stored timestamp, or an earlier row of
the same batch.

latest_seen_nanos = max(prev_total_nanos, batch_latest_nanos.get(redis_key_bin, 0))
if latest_seen_nanos and new_total_nanos <= latest_seen_nanos:
    if progress:
        progress(1)
    continue
batch_latest_nanos[redis_key_bin] = new_total_nanos

redis_key_bin is already computed in the read phase and uniquely identifies
(project, entity_key), so it was free to reuse as the dedup key. The guard had to move
out of the if prev_event_time: block, with prev_total_nanos initialised to 0 so the
max() works when nothing is stored yet — that's the structural part of the diff.

online_write_batch_async had the identical defect and gets the same fix; fixing only the
sync path would have left half the bug in place.

The skip_dedup fast path is deliberately untouched. Its own comment says it is "suitable
for initial loads or append-only pipelines where out-of-order writes are not a concern," so
losing ordering there is the documented tradeoff of enabling it.

One design note worth your input

The issue suggests sorting the batch by timestamp, or reducing to one row per key. Either
would work, but both change behaviour for batches that are already correct — they alter
how many writes get queued in the ascending case and shift the progress() accounting.
I went with the running-max approach because it repairs only the broken case and leaves
every already-working case byte-identical. Happy to switch to sort-or-reduce if you'd
rather have the write-amplification reduction too; it's a small change from here.

Behaviour that is intentionally preserved:

• Rows older than the value already stored are still skipped.
• Ties within a batch keep the first row, matching the existing <= "older or same
  instant" semantics.
• A stored timestamp of epoch 0 still doesn't trigger a skip — the original required a
  truthy prev_total_nanos, and max(0, ...) preserves that.

Note on the labels

The wontfix label on #5163 was applied by stale[bot] as its configured staleLabel,
not by a maintainer declining the report — the timeline shows the bot's comment as the only
activity on the issue. Flagging it so it doesn't read as a prior decision. kind/bug and
priority/p2 are the defaults from bug_report.md. If this is out of scope for other
reasons, I'm glad to hear it.

Which issue(s) this PR fixes:

Fixes #5163

Checks

• I've made sure the tests are passing.
• My commits are signed off (git commit -s)
• My PR title follows conventional commits format

Testing Strategy

• Unit tests
• Integration tests
• Manual tests
• Testing is not required for this change

Misc

Five tests added to sdk/python/tests/unit/infra/online_store/test_redis.py, reusing the
existing MagicMock pipeline idiom, so no Redis server or Docker is needed:

The two that pass before the fix are the load-bearing ones — they're there to show the
change repairs the broken orderings without disturbing the ascending case or the existing
staleness guard.

The pre-fix failure is assert b'\x18\n' == b'\x18\x1e', i.e. int32_val=10 where 30 was
expected — the reporter's "returns 10 instead of 30", reproduced deterministically.

test_redis.py goes 18 → 23 passing. ruff check, ruff format --check and
mypy feast/infra/online_stores/redis.py are all clean.

I ran unit tests only, not the integration suite — it needs cloud credentials I don't have.
Happy to add an integration test if you'd like one for this path.

One small correction to the issue for the record: it describes the code as skipping
"records with timestamps older than what's already been processed in the current batch."
Nothing intra-batch is tracked today, which is exactly why the bug exists — but the symptom
as reported is accurate.