diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 204b08bc..29d76540 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -421,9 +421,11 @@ jobs: run: ./scripts/test-client-compat.sh --filter __none__ --info-manifest --record tmp/client-compat-info.json env: MOON_BIN: ${{ env.CARGO_TARGET_DIR }}/release/moon - # Reports the missing-field set; not yet a gate — info-observability - # owns closing it, and this flips to a hard gate when that task lands. - continue-on-error: true + # HARD GATE as of EC9. Every pinned field is either emitted from a real + # source or waived in info_fields.txt with a recorded reason, and the + # harness refuses an unreasoned waiver (exit 2) as well as a waiver on + # a field Moon has since started emitting — so the waiver list cannot + # silently go stale the way the registry sweep's did. # ── SDK wire-form guards ──────────────────────────────────────────── # The SDK tree had no CI of any kind, which is how five helpers shipped # sending wire forms the server rejects on every call. They live here diff --git a/.gitignore b/.gitignore index 905d6d90..37d61d64 100644 --- a/.gitignore +++ b/.gitignore @@ -106,6 +106,8 @@ tmp/ /target-check-tokio/ /target-check-monoio/ /target-check/ +/target-fu/ +/target-rf/ target-fast/ # CARGO_TARGET_DIR used for the local release-fast build+test loop /target-rf/ diff --git a/CHANGELOG.md b/CHANGELOG.md index 70eef4fa..026371f1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,31 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Added +- **Nine INFO fields a standard monitoring stack reads.** `tcp_port`, `uptime_in_seconds`, + `uptime_in_days`, `aof_last_write_status`, `aof_last_bgrewrite_status`, + `rdb_changes_since_last_save`, `sync_full`, `sync_partial_ok` and `sync_partial_err` are now + emitted, each from a real source rather than a constant: uptime from a start instant captured + before the listener binds, `rdb_changes_since_last_save` from a sharded + keyspace-mutation counter reset at save completion, and the three `sync_*` counters recorded at + the one point in the PSYNC handshake where full-vs-partial is still distinguishable (`PSYNC ? -1` + counts as a full resync the replica ASKED for, not a partial that failed). `tcp_port` reports the + configured listener port, not the port the INFO connection arrived on — behind a container port + map the two differ, and the field exists so a client can hand a peer a reachable address. + The four fields Moon cannot answer truthfully are recorded as waivers with reasons instead of + being emitted: `latest_fork_usec` (Moon never calls `fork(2)`; BGSAVE snapshots in-process), the + two `client_recent_max_*_buffer` high-water marks (untracked, and tracking them means a counter on + every connection read and write), and `used_memory_lua` — which was implemented and then withdrawn + when measurement showed the value the shard can publish is ~2 orders of magnitude below the real + VM footprint on the shipped monoio runtime (80 bytes against tokio's 26183, same `setup_lua_vm`, + ruled out as Cargo feature unification). This keeps the INFO emitter's standing rule intact: a + wrong number on a dashboard is worse than an absent one. + +### Changed +- **The `INFO field coverage` CI step is now a hard gate**, no longer `continue-on-error`. The + pinned-field harness gained a waiver syntax (`field # WAIVED: `) that refuses an + unreasoned waiver at load time (exit 2) and reports a waiver on a field Moon has since started + emitting as a failure — so the list cannot go stale unnoticed the way the registry sweep's did. + - **`CLUSTER SHARDS`, `CLUSTER MYSHARDID`, `READONLY` and `READWRITE`** — the four verbs a cluster-aware client needs to bootstrap. `CLUSTER SHARDS` reports every shard cluster-wide, one entry per master with its replicas, master first; a shard that has lost every live node reports an diff --git a/scripts/client-compat/differ.py b/scripts/client-compat/differ.py index c118fe74..589d960c 100755 --- a/scripts/client-compat/differ.py +++ b/scripts/client-compat/differ.py @@ -612,9 +612,47 @@ def _compare_entry(self, entry: Entry, protocol: str, context: str, return Result(entry.name, protocol, context, r_sent, r_raw, m_raw, verdict, v.divergence, v.detail, entry.expect_diff) + @staticmethod + def _parse_info_manifest(path: str) -> list[tuple[str, str | None]]: + """Pinned INFO fields as (name, waiver_reason | None). + + A line may carry an inline waiver: + + latest_fork_usec # WAIVED: Moon never calls fork(2) + + A waiver is for a field Moon cannot answer TRUTHFULLY -- the INFO + emitter's standing rule is that such a field is omitted rather than + reported as a constant, because a hardcoded zero is indistinguishable + from a healthy server on a dashboard. The waiver records why, so the + omission is a decision on the record instead of a gap. + + A waiver with no reason is rejected at load: an unexplained waiver is + how a real gap gets parked forever. + """ + out: list[tuple[str, str | None]] = [] + with open(path) as f: + for lineno, ln in enumerate(f, 1): + ln = ln.strip() + if not ln or ln.startswith("#"): + continue + name, _, comment = ln.partition("#") + name = name.strip() + comment = comment.strip() + waiver = None + if comment.upper().startswith("WAIVED"): + reason = comment.partition(":")[2].strip() + if not reason: + raise HarnessError( + "ERR_UNREASONED_WAIVER", + f"{path}:{lineno}: '{name}' is waived with no reason. " + f"Write `{name} # WAIVED: `.") + waiver = reason + out.append((name, waiver)) + return out + def _info_coverage(self, rport: int, mport: int) -> list[Result]: - with open(self.cfg.info_manifest) as f: - fields = [ln.strip() for ln in f if ln.strip() and not ln.startswith("#")] + fields = self._parse_info_manifest(self.cfg.info_manifest) rc, mc = RespConn(rport, "resp2"), RespConn(mport, "resp2") try: sent = encode_command(["INFO"]) @@ -628,10 +666,29 @@ def _info_coverage(self, rport: int, mport: int) -> list[Result]: moon_body = (m_node.value or b"").decode("latin1") if m_node else "" redis_body = (r_node.value or b"").decode("latin1") if r_node else "" out = [] - for fname in fields: + for fname, waiver in fields: pat = rf"^{re.escape(fname)}:" in_moon = re.search(pat, moon_body, re.M) is not None in_redis = re.search(pat, redis_body, re.M) is not None + if waiver is not None: + if in_moon: + # Self-invalidating waiver: the field SHIPPED. Leaving the + # waiver in place would keep the manifest green over a + # surface it had stopped needing to excuse -- exactly the + # stale-waiver failure mode the registry sweep hit. + verdict, div = "diff", "value" + detail = (f"'{fname}' is waived ({waiver}) but moon now " + f"EMITS it — delete the waiver, the gap is closed") + out.append(Result(f"info:{fname}", "resp2", "standalone", + sent, r_raw, m_raw, verdict, div, detail)) + continue + # Carry the reason on `waiver_reason`, the field the reporter + # and the JSON record both read — a waiver whose reason prints + # as `None` is indistinguishable from an unreasoned one. + out.append(Result(f"info:{fname}", "resp2", "standalone", sent, + r_raw, m_raw, "waived", None, + f"waived: {waiver}", waiver_reason=waiver)) + continue if not in_redis: # The oracle does not emit it either: the pinned list is wrong, # not Moon. Reporting this as a Moon defect would manufacture a diff --git a/scripts/client-compat/info_fields.txt b/scripts/client-compat/info_fields.txt index a1338261..38a0972d 100644 --- a/scripts/client-compat/info_fields.txt +++ b/scripts/client-compat/info_fields.txt @@ -19,14 +19,14 @@ uptime_in_days # clients connected_clients blocked_clients -client_recent_max_input_buffer -client_recent_max_output_buffer +client_recent_max_input_buffer # WAIVED: per-client input/output buffer high-water marks are not tracked. CLIENT LIST reports qbuf=0 for the same reason. Sampling them means touching a counter on every read and write in the connection hot path, and a hardcoded 0 here would tell an operator hunting a buffer blowup that no client has ever buffered anything. +client_recent_max_output_buffer # WAIVED: see client_recent_max_input_buffer — same untracked source, same hot-path cost. # memory used_memory used_memory_rss used_memory_peak -used_memory_lua +used_memory_lua # WAIVED: the per-shard `mlua` VM's used_memory() is reachable, but on the SHIPPED monoio runtime the value the shard can publish is ~2 orders of magnitude below the real footprint (80 bytes for a VM that reports 26183 under tokio, same setup_lua_vm, same mlua features -- measured both ways, and ruled out as feature unification by building tokio WITH graph+text-index). Emitting it would put a number on a dashboard that is wrong on the runtime that ships. Tracked separately; unwaive when the monoio path samples the VM that actually executes scripts. maxmemory maxmemory_policy mem_fragmentation_ratio @@ -58,7 +58,7 @@ keyspace_hits keyspace_misses pubsub_channels pubsub_patterns -latest_fork_usec +latest_fork_usec # WAIVED: Moon never calls fork(2). BGSAVE snapshots in-process via a copy-on-write epoch cursor, so there is no fork to time. Redis's own value is a fork-duration measurement; emitting 0 would read as 'forks are instant' rather than 'this server does not fork'. # replication role diff --git a/src/admin/metrics_setup.rs b/src/admin/metrics_setup.rs index 4f931bb2..416d3bc7 100644 --- a/src/admin/metrics_setup.rs +++ b/src/admin/metrics_setup.rs @@ -188,6 +188,100 @@ fn total_commands_sum() -> u64 { .sum() } +// ── EC9: keyspace-change counter behind `rdb_changes_since_last_save` ──── +// The "is a save worth doing" signal every backup script reads. Sharded over +// the same padded per-thread slots as the total-commands counter for the same +// reason (a single line would bounce across every shard core at write rate); +// the increment is one relaxed fetch_add on the thread's own line. +// +// Counted at the storage funnels (`Database::set` / `remove` / `get_mut` / +// `clear` / `set_expiry`), not at dispatch: dispatch does not know whether a +// command mutated, and a phf flags lookup on the hot path is exactly the cost +// this codebase's perf invariants forbid. `get_mut` hands out mutable access +// that the caller may or may not use, so the count can run slightly HIGH. +// That direction is deliberate: over-counting says "changes pending" and +// triggers a save that was not needed, while under-counting would tell a +// backup script the dataset was clean when it was not. +static KEYSPACE_CHANGE_COUNTERS: [PaddedCounter; COMMAND_COUNTER_SLOTS] = + [PADDED_COUNTER_ZERO; COMMAND_COUNTER_SLOTS]; +/// Value of the change counter when the last save completed. `rdb_changes_ +/// since_last_save` is the difference; a save that completes concurrently with +/// writes can only make the difference smaller, never negative (saturating). +static KEYSPACE_CHANGES_AT_LAST_SAVE: AtomicU64 = AtomicU64::new(0); + +/// Record one keyspace mutation. Hot path: one relaxed add, no allocation. +#[inline] +pub fn record_keyspace_change() { + COMMAND_COUNTER_SLOT.with(|&slot| { + KEYSPACE_CHANGE_COUNTERS[slot] + .0 + .fetch_add(1, Ordering::Relaxed); + }); +} + +/// Exact sum across all slots. Read paths only (INFO). +fn keyspace_changes_sum() -> u64 { + KEYSPACE_CHANGE_COUNTERS + .iter() + .map(|c| c.0.load(Ordering::Relaxed)) + .sum() +} + +/// Keyspace mutations since the last completed save (INFO `rdb_changes_since_last_save`). +pub fn rdb_changes_since_last_save() -> u64 { + keyspace_changes_sum().saturating_sub(KEYSPACE_CHANGES_AT_LAST_SAVE.load(Ordering::Relaxed)) +} + +/// Mark a save as complete: subsequent changes count from here. +/// +/// Called on SAVE / BGSAVE success, never on failure — a failed save left the +/// dataset unpersisted, so the pending-change count must survive it. +pub fn mark_save_completed() { + KEYSPACE_CHANGES_AT_LAST_SAVE.store(keyspace_changes_sum(), Ordering::Relaxed); +} + +// ── EC9: replica sync counters (INFO `sync_full` / `sync_partial_*`) ───── +// How an operator sees replicas thrashing: a climbing `sync_full` against a +// flat `sync_partial_ok` means partial resync keeps failing and every replica +// reconnect is re-shipping the whole dataset. Plain atomics — these fire once +// per replica handshake, not per command. +static SYNC_FULL: AtomicU64 = AtomicU64::new(0); +static SYNC_PARTIAL_OK: AtomicU64 = AtomicU64::new(0); +static SYNC_PARTIAL_ERR: AtomicU64 = AtomicU64::new(0); + +/// A replica was served a full resync (`+FULLRESYNC`). +#[inline] +pub fn record_sync_full() { + SYNC_FULL.fetch_add(1, Ordering::Relaxed); +} + +/// A replica's `PSYNC ` was satisfied from the backlog. +#[inline] +pub fn record_sync_partial_ok() { + SYNC_PARTIAL_OK.fetch_add(1, Ordering::Relaxed); +} + +/// A replica asked for a partial resync that could not be served. +#[inline] +pub fn record_sync_partial_err() { + SYNC_PARTIAL_ERR.fetch_add(1, Ordering::Relaxed); +} + +/// Full resyncs served since start. +pub fn sync_full() -> u64 { + SYNC_FULL.load(Ordering::Relaxed) +} + +/// Partial resyncs served from the backlog since start. +pub fn sync_partial_ok() -> u64 { + SYNC_PARTIAL_OK.load(Ordering::Relaxed) +} + +/// Partial-resync requests that had to fall back to a full resync. +pub fn sync_partial_err() -> u64 { + SYNC_PARTIAL_ERR.load(Ordering::Relaxed) +} + // ── P6: WAL aggressive reclamation counters (read by P10 INFO emitter) ─── // Incremented by WalWriterV3::recycle_aggressive(). P10 reads these via the // public getters below to populate the `# Reclamation` INFO section. diff --git a/src/command/connection.rs b/src/command/connection.rs index e04f1257..f1aaaf49 100644 --- a/src/command/connection.rs +++ b/src/command/connection.rs @@ -204,6 +204,29 @@ fn run_id() -> &'static str { }) } +/// Process start instant, captured once at startup by [`record_server_start`]. +static SERVER_START: std::sync::OnceLock = std::sync::OnceLock::new(); + +/// Capture the server start instant. Called from `main` before the listener +/// binds, so `uptime_in_seconds` counts from a point that precedes the first +/// client rather than from whenever INFO was first asked. +pub fn record_server_start() { + let _ = SERVER_START.set(std::time::Instant::now()); +} + +/// Seconds since [`record_server_start`]. +/// +/// Falls back to initialising on first read, which keeps a unit test or an +/// embedded harness that never calls `record_server_start` reporting a +/// monotonic uptime instead of a constant — the only wrong answer here is one +/// that never advances, because that is what makes a crash loop invisible. +fn server_uptime_secs() -> u64 { + SERVER_START + .get_or_init(std::time::Instant::now) + .elapsed() + .as_secs() +} + /// Build the full INFO payload with every section present. /// /// Callers must pass this through [`crate::command::info_sections::finalize`], @@ -238,6 +261,17 @@ fn info_raw(db: &Database, facts: &InstanceFacts) -> String { sections.push_str("cluster_enabled:0\r\n"); } let _ = write!(sections, "process_id:{}\r\n", std::process::id()); + // The port this instance LISTENS on, not the port this connection arrived + // on. Behind a container port map or a proxy they differ, and a client + // handing a peer the arrival port would send it somewhere unreachable. + let _ = write!(sections, "tcp_port:{}\r\n", facts.tcp_port); + // Uptime is the field every restart-detector keys on: a drop to near zero + // is how a dashboard learns the process died. Sourced from the start + // instant captured before the listener binds, so it can never read as a + // constant. + let uptime_secs = server_uptime_secs(); + let _ = write!(sections, "uptime_in_seconds:{uptime_secs}\r\n"); + let _ = write!(sections, "uptime_in_days:{}\r\n", uptime_secs / 86_400); let _ = write!(sections, "os:{}\r\n", std::env::consts::OS); let _ = write!( sections, @@ -401,9 +435,12 @@ fn info_raw(db: &Database, facts: &InstanceFacts) -> String { }; sections.push_str(&format!( "loading:0\r\n\ + rdb_changes_since_last_save:{}\r\n\ rdb_bgsave_in_progress:{}\r\n\ rdb_last_save_time:{}\r\n\ rdb_last_bgsave_status:{}\r\n\ + aof_last_write_status:{}\r\n\ + aof_last_bgrewrite_status:{}\r\n\ aof_enabled:{}\r\n\ aof_rewrite_in_progress:{}\r\n\ aof_base_size:{}\r\n\ @@ -419,6 +456,10 @@ fn info_raw(db: &Database, facts: &InstanceFacts) -> String { spill_failed_reinserted:{}\r\n\ spill_completion_superseded:{}\r\n\ spill_last_heartbeat_ms:{}\r\n", + // Keyspace mutations since the last COMPLETED save — the "is a save + // worth doing" signal a backup script reads. A failed save does not + // reset it: the dataset is still unpersisted. + crate::admin::metrics_setup::rdb_changes_since_last_save(), if crate::command::persistence::SAVE_IN_PROGRESS.load(std::sync::atomic::Ordering::Relaxed) { 1 @@ -433,6 +474,20 @@ fn info_raw(db: &Database, facts: &InstanceFacts) -> String { } else { "err" }, + // Redis parity names for the two AOF statuses stock tooling + // string-matches. `aof_last_write_status` shares its source with + // Moon's own `aof_last_append_status` below — the same fact under the + // name a redis-py/ioredis health check actually looks for. + if crate::persistence::aof::AOF_LAST_APPEND_OK.load(std::sync::atomic::Ordering::Relaxed) { + "ok" + } else { + "err" + }, + if crate::persistence::aof::AOF_REWRITE_LAST_OK.load(std::sync::atomic::Ordering::Relaxed) { + "ok" + } else { + "err" + }, u8::from(aof_enabled), u8::from( crate::command::persistence::AOF_REWRITE_IN_PROGRESS @@ -533,6 +588,9 @@ fn info_raw(db: &Database, facts: &InstanceFacts) -> String { total_net_input_bytes:{}\r\n\ total_net_output_bytes:{}\r\n\ instantaneous_ops_per_sec:{}\r\n\ + sync_full:{}\r\n\ + sync_partial_ok:{}\r\n\ + sync_partial_err:{}\r\n\ pubsub_channels:{}\r\n\ pubsub_patterns:{}\r\n", crate::admin::metrics_setup::keyspace_hits(), @@ -543,6 +601,12 @@ fn info_raw(db: &Database, facts: &InstanceFacts) -> String { crate::admin::metrics_setup::total_net_input_bytes(), crate::admin::metrics_setup::total_net_output_bytes(), crate::admin::metrics_setup::instantaneous_ops_per_sec(), + // Replica-sync health: a climbing `sync_full` against a flat + // `sync_partial_ok` means partial resync keeps failing and every + // replica reconnect re-ships the whole dataset. + crate::admin::metrics_setup::sync_full(), + crate::admin::metrics_setup::sync_partial_ok(), + crate::admin::metrics_setup::sync_partial_err(), facts.pubsub_channels, facts.pubsub_patterns, ); @@ -655,6 +719,13 @@ pub struct InstanceFacts { pub pubsub_channels: usize, /// Distinct subscribed patterns, across all shards. pub pubsub_patterns: usize, + /// The port this instance's listener is bound to (`--port`). + /// + /// Deliberately the CONFIGURED port and not the local port of the socket + /// INFO arrived on: behind a container port map or a proxy the two differ, + /// and `tcp_port` exists so a client can hand a peer an address that + /// actually reaches this server. + pub tcp_port: u16, } /// As [`info_with_keyspace_and_replication`], plus the instance-wide facts diff --git a/src/command/persistence.rs b/src/command/persistence.rs index a9c34db7..11ba5bd0 100644 --- a/src/command/persistence.rs +++ b/src/command/persistence.rs @@ -101,6 +101,8 @@ pub fn bgsave_start(db: SharedDatabases, dir: String, dbfilename: String) -> Fra Ok(()) => { info!("Background RDB save completed: {}", path.display()); BGSAVE_LAST_STATUS.store(true, Ordering::Relaxed); + // A completed save is the reset point for `rdb_changes_since_last_save`. + crate::admin::metrics_setup::mark_save_completed(); LAST_SAVE_TIME.store( std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) @@ -126,6 +128,8 @@ pub fn bgsave_start(db: SharedDatabases, dir: String, dbfilename: String) -> Fra Ok(()) => { info!("Background RDB save completed: {}", path.display()); BGSAVE_LAST_STATUS.store(true, Ordering::Relaxed); + // A completed save is the reset point for `rdb_changes_since_last_save`. + crate::admin::metrics_setup::mark_save_completed(); LAST_SAVE_TIME.store( std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) @@ -205,6 +209,8 @@ pub fn bgsave_shard_done(success: bool) { if prev == 1 { // Last shard to finish SAVE_IN_PROGRESS.store(false, Ordering::SeqCst); + // A completed save is the reset point for `rdb_changes_since_last_save`. + crate::admin::metrics_setup::mark_save_completed(); LAST_SAVE_TIME.store( std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) @@ -372,6 +378,8 @@ pub fn handle_save(db: &SharedDatabases, dir: &str, dbfilename: &str) -> Frame { let path = PathBuf::from(dir).join(dbfilename); match rdb::save_from_snapshot(&snapshot, &path) { Ok(()) => { + // A completed save is the reset point for `rdb_changes_since_last_save`. + crate::admin::metrics_setup::mark_save_completed(); LAST_SAVE_TIME.store( std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) diff --git a/src/main.rs b/src/main.rs index 35f06a53..2d6b98c6 100644 --- a/src/main.rs +++ b/src/main.rs @@ -73,6 +73,13 @@ fn main() -> anyhow::Result<()> { // into ONE re-spawn with ONE conf string (see `build_malloc_conf`). malloc_respawn::maybe_respawn_with_memory_overrides()?; + // EC9: capture the start instant for INFO `uptime_in_seconds`. Placed + // immediately after the (at most one) malloc re-exec so uptime measures + // THIS process image, not the one that exec'd away — and before every + // subsystem, so a slow index reload counts as uptime rather than + // disappearing from it. + moon::command::connection::record_server_start(); + // Block SIGTERM in the main thread BEFORE any child threads are spawned // (TLS reload thread, Prometheus admin thread, ctrlc internal thread, // shard threads). All child threads inherit the blocked mask, so SIGTERM diff --git a/src/persistence/aof/mod.rs b/src/persistence/aof/mod.rs index becee119..258258e2 100644 --- a/src/persistence/aof/mod.rs +++ b/src/persistence/aof/mod.rs @@ -146,6 +146,20 @@ pub fn record_everysec_fsync_result(writer_idx: usize, ok: bool) { pub static AOF_LAST_APPEND_OK: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(true); +/// Whether the last AOF rewrite (`BGREWRITEAOF`, or the #433 auto-rewrite) +/// completed. Surfaces as INFO `aof_last_bgrewrite_status` — the field an +/// operator reads after a disk incident to learn whether the rewrite that was +/// supposed to reclaim the backlog actually finished. +/// +/// `true` before any rewrite has run: Redis reports `ok` on a fresh instance +/// too, and reporting `err` for "never attempted" would page someone for a +/// rewrite that was never asked for. Set on every rewrite completion, both +/// directions — a later success clears an earlier failure, because unlike +/// `AOF_LAST_APPEND_OK` (which latches a permanently-missing record) a +/// successful rewrite genuinely restores the invariant. +pub static AOF_REWRITE_LAST_OK: std::sync::atomic::AtomicBool = + std::sync::atomic::AtomicBool::new(true); + /// Reason-DELs (eviction/expiry) dropped at the writer channel (#452.4). /// Strictly worse than a dropped client write: replay RESURRECTS a key the /// server told clients was gone. Kept as a dedicated counter so a non-zero @@ -464,6 +478,7 @@ impl PerShardRewriteCoord { // Publish BEFORE clearing the in-progress flag so every blocked // writer wakes and reopens onto the (still-authoritative) old gen. self.publish_outcome(self.old_seq); + AOF_REWRITE_LAST_OK.store(false, Ordering::SeqCst); crate::command::persistence::AOF_REWRITE_IN_PROGRESS.store(false, Ordering::SeqCst); return; } @@ -481,6 +496,7 @@ impl PerShardRewriteCoord { // writers must roll back to it (their phase-6 reopen targeted // new_seq, which never committed). self.publish_outcome(self.old_seq); + AOF_REWRITE_LAST_OK.store(false, Ordering::SeqCst); crate::command::persistence::AOF_REWRITE_IN_PROGRESS.store(false, Ordering::SeqCst); return; } @@ -540,6 +556,7 @@ impl PerShardRewriteCoord { // new_seq in phase 6, so the barrier is a no-op for them — but it must // still unblock them. self.publish_outcome(self.new_seq); + AOF_REWRITE_LAST_OK.store(true, Ordering::SeqCst); crate::command::persistence::AOF_REWRITE_IN_PROGRESS.store(false, Ordering::SeqCst); } } diff --git a/src/replication/master.rs b/src/replication/master.rs index a29a9ca1..786a2754 100644 --- a/src/replication/master.rs +++ b/src/replication/master.rs @@ -79,16 +79,29 @@ pub async fn handle_psync_inline_single_shard( }; // Decide full vs partial resync against the single-shard backlog. + // + // EC9: the three INFO `sync_*` counters are recorded here, at the one + // point where the distinction is still visible. `PSYNC ? -1` is a replica + // ASKING for a full resync, not a partial resync that failed — counting it + // as `sync_partial_err` would make a healthy first-time replica look like + // a backlog problem. Only a replica that offered a replid+offset and was + // refused counts as a partial-resync failure. let decision = if client_offset < 0 { + crate::admin::metrics_setup::record_sync_full(); PsyncDecision::FullResync } else if client_repl_id != repl_id && client_repl_id != repl_id2 { + crate::admin::metrics_setup::record_sync_partial_err(); + crate::admin::metrics_setup::record_sync_full(); PsyncDecision::FullResync } else { let off = client_offset as u64; let g = backlog_slot.lock(); if g.as_ref().is_some_and(|b| b.contains_offset(off)) { + crate::admin::metrics_setup::record_sync_partial_ok(); PsyncDecision::PartialResync { from_offset: off } } else { + crate::admin::metrics_setup::record_sync_partial_err(); + crate::admin::metrics_setup::record_sync_full(); PsyncDecision::FullResync } }; diff --git a/src/server/conn/handler_monoio/dispatch.rs b/src/server/conn/handler_monoio/dispatch.rs index 5cf51baf..ab7c738c 100644 --- a/src/server/conn/handler_monoio/dispatch.rs +++ b/src/server/conn/handler_monoio/dispatch.rs @@ -737,6 +737,7 @@ pub(super) async fn try_handle_info( let pubsub_facts = conn_cmd::InstanceFacts { pubsub_channels, pubsub_patterns, + tcp_port: ctx.config_port, }; let resp_frame = crate::shard::slice::with_shard_db(conn.selected_db, |db| { conn_cmd::info_with_facts(db, cmd_args, &keyspace, real_repl.as_deref(), &pubsub_facts) diff --git a/src/server/conn/handler_sharded/dispatch.rs b/src/server/conn/handler_sharded/dispatch.rs index 38a4c1a5..9cb28350 100644 --- a/src/server/conn/handler_sharded/dispatch.rs +++ b/src/server/conn/handler_sharded/dispatch.rs @@ -372,6 +372,7 @@ pub(super) async fn try_handle_info( let pubsub_facts = conn_cmd::InstanceFacts { pubsub_channels, pubsub_patterns, + tcp_port: ctx.config_port, }; let resp_frame = crate::shard::slice::with_shard_db(conn.selected_db, |db| { conn_cmd::info_with_facts(db, cmd_args, &keyspace, real_repl.as_deref(), &pubsub_facts) diff --git a/src/server/conn/handler_single.rs b/src/server/conn/handler_single.rs index 26e41a4d..eb411ae6 100644 --- a/src/server/conn/handler_single.rs +++ b/src/server/conn/handler_single.rs @@ -1088,6 +1088,7 @@ pub async fn handle_connection( conn_cmd::InstanceFacts { pubsub_channels: reg.active_channels(None).len(), pubsub_patterns: reg.pattern_names().len(), + tcp_port: config.port, } }; let guard = db[conn.selected_db].read(); diff --git a/src/storage/db/kv_ops.rs b/src/storage/db/kv_ops.rs index 792bc6ee..db519683 100644 --- a/src/storage/db/kv_ops.rs +++ b/src/storage/db/kv_ops.rs @@ -292,6 +292,10 @@ impl Database { /// Optimized: immutable check for expiry (rare path), then single get_mut /// for LRU touch + return. Reduces from 3 lookups to 2 for non-expired keys. pub fn get_mut(&mut self, key: &[u8]) -> Option<&mut Entry> { + // Mutable access is a write intent; see `record_keyspace_change` + // for why counting the intent (rather than the mutation) is the + // safe direction for `rdb_changes_since_last_save`. + crate::admin::metrics_setup::record_keyspace_change(); let now = self.cached_now; let now_ms = self.cached_now_ms; // Immutable check for expiry (avoids get_mut + remove + get_mut triple lookup) @@ -315,6 +319,7 @@ impl Database { /// and miss paths. The old `get_mut` + `insert` pattern ran two probes on /// miss (PERF-08). pub fn set(&mut self, key: Bytes, entry: Entry) { + crate::admin::metrics_setup::record_keyspace_change(); // An overwrite makes any in-flight spill payload for this key stale. // Retiring the record here stops its completion publishing the OLD // value into `cold_index`, where it would sit as a shadow behind the @@ -414,6 +419,7 @@ impl Database { /// before loading the authoritative base RDB + incr log. Without this, /// non-idempotent commands from pre-existing state would be double-applied. pub fn clear(&mut self) { + crate::admin::metrics_setup::record_keyspace_change(); self.data = DashTable::new(); self.used_memory = 0; self.maybe_has_expiring_keys = false; @@ -497,6 +503,7 @@ impl Database { /// key returns `None` (no in-RAM entry exists); callers that must COUNT /// cold-only removals (DEL/UNLINK) use [`Self::remove_counting_cold`]. pub fn remove(&mut self, key: &[u8]) -> Option { + crate::admin::metrics_setup::record_keyspace_change(); let _ = self.remove_cold_only(key); self.remove_hot(key) } @@ -509,6 +516,7 @@ impl Database { /// removed hot entry, when present, is also returned so UNLINK can /// size its async-drop decision. pub fn remove_counting_cold(&mut self, key: &[u8]) -> (bool, Option) { + crate::admin::metrics_setup::record_keyspace_change(); let now_ms = self.cached_now_ms; let cold_alive = self .cold_index @@ -878,6 +886,7 @@ impl Database { /// Performs lazy expiry check first. Returns `false` if the key does not /// exist (or has already expired). Pass 0 to remove expiry. pub fn set_expiry(&mut self, key: &[u8], expires_at_ms: u64) -> bool { + crate::admin::metrics_setup::record_keyspace_change(); let now_ms = self.cached_now_ms; if Self::check_expired(&self.data, key, now_ms) { if let Some(entry) = self.data.remove(key) { diff --git a/tests/info_observability.rs b/tests/info_observability.rs index 2d325d30..20136ff7 100644 --- a/tests/info_observability.rs +++ b/tests/info_observability.rs @@ -44,9 +44,21 @@ fn spawn_moon(shards: &str) -> Moon { spawn_moon_in(shards, None) } +/// A server with persistence ON. BGSAVE only completes when a durability +/// backstop is configured — with `appendonly no` and no `--save`, Moon logs +/// "BGSAVE triggered" and the snapshot epoch is never advanced, so a test that +/// waits for a save on the default spawner waits forever. +fn spawn_moon_persistent(shards: &str) -> Moon { + spawn_moon_with(shards, None, &["--appendonly", "yes"]) +} + /// `dir` lets a restart test reuse the SAME data dir, which is the only way /// to prove run_id changes for reasons other than a fresh dataset. fn spawn_moon_in(shards: &str, dir: Option) -> Moon { + spawn_moon_with(shards, dir, &["--appendonly", "no"]) +} + +fn spawn_moon_with(shards: &str, dir: Option, extra: &[&str]) -> Moon { let bin = std::path::PathBuf::from(env!("CARGO_BIN_EXE_moon")); let fixed_dir = dir.clone(); let (child, port) = common::spawn_listening(|port| { @@ -62,13 +74,12 @@ fn spawn_moon_in(shards: &str, dir: Option) -> Moon { shards, "--admin-port", "0", - "--appendonly", - "no", "--disk-free-min-pct", "0", "--dir", tmp_dir.to_str().unwrap(), ]) + .args(extra) .stdout(Stdio::null()) .stderr( std::fs::File::create(tmp_dir.join("moon.stderr")).expect("create moon stderr log"), @@ -502,3 +513,177 @@ fn io13_pubsub_counts_are_instance_wide() { "one subscribed pattern must be visible instance-wide" ); } + +// --------------------------------------------------------------------------- +// io14..io18 — the ten INFO fields the pinned client manifest reads but Moon +// did not answer. Each is backed by a real source; a field Moon cannot answer +// truthfully is waived in scripts/client-compat/info_fields.txt with a reason, +// not emitted as a constant. See `# Server`/`# Stats` in command/connection.rs. +// --------------------------------------------------------------------------- + +/// `tcp_port` is how a client that reached the server via a proxy, a container +/// port map, or a sentinel handoff learns the port to hand to a peer. Reporting +/// the port the connection arrived on would be wrong behind a NAT — this must +/// be the listener's own configured port. +#[test] +fn io14_tcp_port_is_the_configured_listener_port() { + let m = spawn_moon("1"); + let mut c = Conn::open(m.port); + let reply = c.send(&["INFO", "server"]); + let got = field(&reply, "tcp_port") + .unwrap_or_else(|| panic!("INFO server has no tcp_port field; got:\n{reply}")); + assert_eq!( + got, + m.port.to_string(), + "tcp_port must be the port this instance listens on ({}), not {got}", + m.port + ); +} + +/// `uptime_in_seconds` is the field every restart-detector keys on: a drop to +/// near zero is how a dashboard learns the process died. A constant, or a value +/// that never advances, makes a crash-looping server indistinguishable from a +/// healthy one. +#[test] +fn io15_uptime_advances_and_days_agree() { + let m = spawn_moon("1"); + let mut c = Conn::open(m.port); + + let first = field(&c.send(&["INFO", "server"]), "uptime_in_seconds") + .unwrap_or_else(|| panic!("INFO server has no uptime_in_seconds")) + .parse::() + .expect("uptime_in_seconds must be an integer"); + assert!( + first < 60, + "a just-spawned server reported uptime_in_seconds={first} — the start \ + instant is not being captured at startup" + ); + + std::thread::sleep(Duration::from_millis(1600)); + let reply = c.send(&["INFO", "server"]); + let second = field(&reply, "uptime_in_seconds") + .expect("uptime_in_seconds") + .parse::() + .expect("integer"); + assert!( + second > first, + "uptime_in_seconds did not advance across 1.6s ({first} -> {second}); \ + a frozen uptime hides a restart from every monitoring agent" + ); + + let days = field(&reply, "uptime_in_days") + .expect("uptime_in_days") + .parse::() + .expect("integer"); + assert_eq!( + days, + second / 86_400, + "uptime_in_days must be uptime_in_seconds/86400, not an independent counter" + ); +} + +/// The two AOF status fields are the ones an operator reads after a disk +/// incident. Redis reports `ok`/`err`; anything else breaks the parse in +/// stock tooling. With AOF off they must still report a defined status. +#[test] +fn io17_aof_status_fields_are_ok_or_err() { + let m = spawn_moon("1"); + let mut c = Conn::open(m.port); + let reply = c.send(&["INFO", "persistence"]); + + for name in ["aof_last_write_status", "aof_last_bgrewrite_status"] { + let got = field(&reply, name) + .unwrap_or_else(|| panic!("INFO persistence has no {name}; got:\n{reply}")); + assert!( + got == "ok" || got == "err", + "{name} must be `ok` or `err` (Redis parity — tooling string-matches \ + these), got {got:?}" + ); + } +} + +/// `rdb_changes_since_last_save` is the "is a save worth doing" signal. It must +/// rise with writes and reset when a save completes; a field pinned at 0 tells +/// a backup script there is nothing to persist. +#[test] +fn io18_rdb_changes_tracks_writes_and_resets_on_save() { + let m = spawn_moon_persistent("1"); + let mut c = Conn::open(m.port); + + let base = field( + &c.send(&["INFO", "persistence"]), + "rdb_changes_since_last_save", + ) + .unwrap_or_else(|| panic!("INFO persistence has no rdb_changes_since_last_save")) + .parse::() + .expect("integer"); + + for i in 0..25 { + c.send(&["SET", &format!("io18:{i}"), "v"]); + } + let after_writes = field( + &c.send(&["INFO", "persistence"]), + "rdb_changes_since_last_save", + ) + .expect("field") + .parse::() + .expect("integer"); + assert!( + after_writes >= base + 25, + "25 SETs advanced rdb_changes_since_last_save by {} (expected >= 25) — \ + the counter is not fed by the write path", + after_writes - base + ); + + // BGSAVE, not SAVE: SAVE is refused in sharded mode, and Moon spawns + // every instance sharded. BGSAVE returns before the save finishes, so the + // reset must be observed by polling `rdb_bgsave_in_progress` rather than + // read immediately — reading too early would pass for the wrong reason + // (the counter simply had not been reset yet). + let saved = c.send(&["BGSAVE"]); + assert!( + saved.starts_with('+'), + "BGSAVE failed, test proves nothing: {saved}" + ); + let deadline = Instant::now() + Duration::from_secs(20); + let mut after_save = after_writes; + while Instant::now() < deadline { + let reply = c.send(&["INFO", "persistence"]); + let in_progress = field(&reply, "rdb_bgsave_in_progress").unwrap_or_default(); + after_save = field(&reply, "rdb_changes_since_last_save") + .expect("field") + .parse::() + .expect("integer"); + if in_progress == "0" && after_save < after_writes { + break; + } + std::thread::sleep(Duration::from_millis(100)); + } + assert!( + after_save < after_writes, + "a completed SAVE did not reset rdb_changes_since_last_save \ + ({after_writes} -> {after_save}); the field never returns to a \ + 'nothing to persist' state" + ); +} + +/// The three `sync_*` counters are how an operator sees replicas thrashing: +/// a climbing `sync_full` means partial resync keeps failing. On a standalone +/// master with no replicas they must be present and zero — present, because a +/// missing field breaks the scrape; zero, because nothing has synced. +#[test] +fn io19_sync_counters_present_and_zero_without_replicas() { + let m = spawn_moon("1"); + let mut c = Conn::open(m.port); + let reply = c.send(&["INFO", "stats"]); + for name in ["sync_full", "sync_partial_ok", "sync_partial_err"] { + let got = field(&reply, name) + .unwrap_or_else(|| panic!("INFO stats has no {name}; got:\n{reply}")) + .parse::() + .unwrap_or_else(|_| panic!("{name} must be an integer")); + assert_eq!( + got, 0, + "{name} is {got} on an instance no replica ever contacted" + ); + } +}