feat(meta): implement distributed semaphore based on meta-service (#17651)

* docs: update meta-service change-log

* feat(meta): implement distributed semaphore based on meta-service

This commit introduces a distributed semaphore implementation that provides
reliable resource management across distributed systems. The semaphore is
built on top of meta-service and offers:

- Fair queueing mechanism using sequence numbers
- Automatic lease management with TTL
- Atomic operations for consistency
- Real-time updates via meta-service watch API

Usage:

```rust
let client = MetaGrpcClient::try_create(/*..*/);
let acquired_guard = Semaphore::new_acquired(
        client,
        "your/semaphore/name/in/meta/service",
        2,                          // capacity: 2 acquired at most
        "id11",                     // ID of this acquirer
        Duration::from_secs(3)      // lease time
).await?;

acquired_guard.await;
// Released
```

The implementation uses a queue-based structure with sequence numbers to
ensure fair ordering of resource requests. It handles distributed
coordination through meta-service, providing automatic lease extension
and cleanup of stale entries.

Key components:
- Semaphore: Main entry point for semaphore operations
- Acquirer: Handles the acquisition process
- AcquiredGuard: Manages semaphore lifecycle
- Queue-based storage with sequence numbers

Note: Current version sets capacity per client. Future versions will
store capacity in meta key for global consistency.

* chore: add example

* chore: fix fmt

* chore: fix type changes

* refactor: rename semaphore key|entry to permit key|entry

* chore: remove unused deps

* chore: fix lint

* chore: fix display ut for permit

* chore: fix lint

* chore: fix ut

* refactor: rename AcquiredGuard to Permit

* refactor: refine acquirer doc

* refactor: rename SemaphoreEvent to PermitEvent

* chore: rename

* chore: refine path

* update meta-service compat doc

Author: drmingdrmer

Cargo.lock

@@ -105,6 +105,7 @@ members = [
     "src/meta/ee",
     "src/meta/proto-conv",
     "src/meta/protos",
+    "src/meta/semaphore",
     "src/meta/service",
     "tests/sqllogictests",
     "src/tests/sqlsmith",
@@ -146,6 +147,7 @@ databend-common-meta-embedded = { path = "src/meta/embedded" }
 databend-common-meta-kvapi = { path = "src/meta/kvapi" }
 databend-common-meta-process = { path = "src/meta/process" }
 databend-common-meta-raft-store = { path = "src/meta/raft-store" }
+databend-common-meta-semaphore = { path = "src/meta/semaphore" }
 databend-common-meta-sled-store = { path = "src/meta/sled-store" }
 databend-common-meta-stoerr = { path = "src/meta/stoerr" }
 databend-common-meta-store = { path = "src/meta/store" }
@@ -274,6 +276,7 @@ chrono = { version = "0.4.31", features = ["serde"] }
 chrono-tz = { version = "0.8", features = ["serde"] }
 cidr = { version = "0.2.2" }
 clap = { version = "4.4.2", features = ["derive"] }
+codeq = { version = "0.5.2" }
 comfy-table = "7"
 convert_case = "0.6.0"
 cookie = "0.18.1"

Cargo.toml

@@ -77,22 +77,29 @@ To add a new feature(add new type or update an type), the developer should do:
 
 The following is an illustration of the latest query-meta compatibility:
 
-| `Meta\Query`       | [0.9.41, 1.1.34) | [1.1.34, 1.2.287) | [1.2.287, 1.2.361) | [1.2.361, +∞) |
-|:-------------------|:-----------------|:---------------|:-----------|:-----------|
-| [0.8.30, 0.8.35)   | ❌                | ❌              | ❌          |❌          |
-| [0.8.35, 0.9.23)   | ✅                | ❌              | ❌          |❌          |
-| [0.9.23, 0.9.42)   | ✅                | ❌              | ❌          |❌          |
-| [0.9.42, 1.1.32)   | ✅                | ❌              | ❌          |❌          |
-| [1.1.32, 1.2.63)   | ✅                | ✅              | ❌          |❌          |
-| [1.2.63, 1.2.226)  | ✅                | ✅              | ❌          |❌          |
-| [1.2.226, 1.2.258) | ✅                | ✅              | ✅          |❌          |
-| [1.2.258, 1.2.663) | ✅                | ✅              | ✅          |✅          |
-| [1.2.663, +∞)      | ❌                | ❌              | ✅          |✅          |
+`[, a.b.c)` denotes the range of versions from previous version(on the left column)(inclusive)
+upto `a.b.c` (exclusive).
+
+TODO: xx is the version in which semaphore is added. update it when merged.
+
+| `Meta\Query`       | 1.1.34) | [, 1.2.287) | [, 1.2.361) | [, xx) | [, +∞)           |
+|:-------------------|:--------|:------------|:------------|:-------|:-----------------|
+| [0.8.30, 0.8.35)   |         | ❌           | ❌           | ❌      | ❌                |
+| [0.8.35, 0.9.23)   |         | ❌           | ❌           | ❌      | ❌                |
+| [0.9.23, 0.9.42)   |         | ❌           | ❌           | ❌      | ❌                |
+| [0.9.42, 1.1.32)   |         | ❌           | ❌           | ❌      | ❌                |
+| [1.1.32, 1.2.63)   |         | ✅           | ❌           | ❌      | ❌                |
+| [1.2.63, 1.2.226)  |         | ✅           | ❌           | ❌      | ❌                |
+| [1.2.226, 1.2.258) |         | ✅           | ✅           | ❌      | ❌                |
+| [1.2.258, 1.2.663) |         | ✅           | ✅           | ✅      | ✅(no semaphore)  |
+| [1.2.663, 1.2.677) |         | ❌           | ✅           | ✅      | ✅(no semaphore)  |
+| [1.2.677, +∞)      |         | ❌           | ✅           | ✅      | ✅                |
 
 History versions that are not included in the above chart:
 
 - Query `[0.7.59, 0.8.80)` is compatible with Meta `[0.8.30, 0.9.23)`.
 - Query `[0.8.80, 0.9.41)` is compatible with Meta `[0.8.35, 0.9.42)`.
+- Query `[0.9.41, 1.1.34)` is compatible with Meta `[0.8.35, 1.2.663)`.
 
 
 ## Compatibility between databend-meta

src/meta/README.md

@@ -58,6 +58,7 @@ use databend_common_meta_types::protobuf::WatchRequest;
 use databend_common_meta_types::protobuf::WatchResponse;
 use databend_common_meta_types::ConnectionError;
 use databend_common_meta_types::GrpcConfig;
+use databend_common_meta_types::InvalidArgument;
 use databend_common_meta_types::MetaClientError;
 use databend_common_meta_types::MetaError;
 use databend_common_meta_types::MetaHandshakeError;
@@ -986,6 +987,26 @@ impl MetaGrpcClient {
         debug!("{}: handle watch request: {:?}", self, watch_request);
 
         let mut client = self.get_established_client().await?;
+
+        if watch_request.initial_flush {
+            let server_version = client.server_protocol_version();
+            let least_server_version = 1002677;
+
+            if server_version < least_server_version {
+                let err = format!(
+                    "WatchRequest::initial_flush requires databend-meta is at least {}, but: {}",
+                    least_server_version, server_version
+                );
+
+                error!("{}", err);
+
+                return Err(InvalidArgument::new(
+                    AnyError::error("databend-meta version too low"),
+                    err,
+                )
+                .into());
+            }
+        }
         let res = client.watch(watch_request).await?;
         Ok(res.into_inner())
     }

src/meta/client/src/grpc_client.rs

@@ -113,14 +113,21 @@ pub static METACLI_COMMIT_SEMVER: LazyLock<Version> = LazyLock::new(|| {
 /// - 2024-11-23: since 1.2.663
 ///   👥 client: remove use of `Operation::AsIs`
 ///
-/// - 2024-12-1*: since 1.2.*
+/// - 2024-12-16: since 1.2.674
 ///   🖥 server: add `txn_condition::Target::KeysWithPrefix`,
 ///              to support matching the key count by a prefix.
 ///
-/// - 2024-12-1*: since 1.2.*
-///   🖥 server: add `TxnRequest::condition_tree`,
-///              to specify a complex bool expression.
+/// - 2024-12-20: since 1.2.676
+///   🖥 server: add `TxnRequest::operations`,
+///              to specify a complex bool expression and corresponding operations
 ///
+/// - 2024-12-26: since 1.2.677
+///   🖥 server: add `WatchRequest::initial_flush`,
+///              to let watch stream flush all keys in a range at the beginning.
+///
+/// - 2025-03-28: since TODO: add version when merged.
+///   👥 client: semaphore(watch) requires `WatchRequest::initial_flush`(`1,2.677`),
+///              other RPC does not require `1.2.677`, requires only `1.2.259`.
 ///
 /// Server feature set:
 /// ```yaml

src/meta/client/src/lib.rs

@@ -0,0 +1,33 @@
+[package]
+name = "databend-common-meta-semaphore"
+description = "A semaphore impl based on distributed meta-service"
+version = { workspace = true }
+authors = { workspace = true }
+license = { workspace = true }
+publish = { workspace = true }
+edition = { workspace = true }
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[lib]
+doctest = false
+test = true
+
+[dependencies]
+codeq = { workspace = true }
+databend-common-base = { workspace = true }
+databend-common-meta-client = { workspace = true }
+databend-common-meta-kvapi = { workspace = true }
+databend-common-meta-types = { workspace = true }
+futures = { workspace = true }
+log = { workspace = true }
+thiserror = { workspace = true }
+tokio = { workspace = true }
+tonic = { workspace = true }
+
+[dev-dependencies]
+anyhow = { workspace = true }
+pretty_assertions = { workspace = true }
+
+[lints]
+workspace = true

src/meta/semaphore/Cargo.toml

@@ -0,0 +1,87 @@
+# Databend Common Meta Semaphore
+
+A distributed semaphore implementation based on meta-service, providing reliable resource management across distributed systems.
+
+## Overview
+
+This crate implements a distributed semaphore using a meta-service as the underlying storage and coordination mechanism. It provides a robust way to manage and coordinate access to limited resources across distributed systems.
+
+## Features
+
+- **Distributed Coordination**: Uses meta-service for reliable distributed coordination
+- **Fair Queueing**: Implements a fair queueing mechanism based on sequence numbers
+- **Automatic Lease Management**: Handles TTL and lease extension automatically
+- **Atomic Operations**: Ensures consistency through atomic operations on meta-service
+- **Event-based Updates**: Real-time updates through meta-service watch API
+
+## Key Components
+
+### Semaphore Structure
+
+```text
+<prefix>/meta          -> {capacity: 10}  // Planned for future versions
+<prefix>/queue/<seq_1> -> {id: "<id_1>", value: 1}
+<prefix>/queue/<seq_2> -> {id: "<id_2>", value: 2}
+<prefix>/queue/<seq_3> -> {id: "<id_3>", value: 1}
+<prefix>/seq_generator -> {}
+```
+
+- `<prefix>`: User-defined string to identify a semaphore instance
+- `queue/*`: Contains semaphore entries with sequence numbers
+- `seq_generator`: Generates globally unique sequence numbers
+- Each entry contains:
+  - `id`: User-defined identifier
+  - `value`: Resource amount consumed
+
+### Main Types
+
+- `Semaphore`: The main entry point for semaphore operations
+- `Acquirer`: Handles the semaphore acquisition process
+- `AcquiredGuard`: Manages the lifecycle of an acquired semaphore
+- `SemaphoreEntry`: Represents a semaphore entry in the queue
+- `SemaphoreKey`: Defines the key structure for semaphore entries
+
+## Usage
+
+```rust
+let client = MetaGrpcClient::try_create(/*..*/);
+let acquired_guard = Semaphore::new_acquired(
+        client,
+        "your/semaphore/name/in/meta/service",
+        2,                          // capacity: 2 acquired at most
+        "id11",                     // ID of this acquirer
+        Duration::from_secs(3)      // lease time
+).await?;
+
+acquired_guard.await;
+// Released
+```
+
+## Implementation Details
+
+### Acquisition Process
+
+1. Client obtains a new sequence number from `seq_generator`
+2. Creates a `SemaphoreEntry` and inserts it at `queue/<seq>`
+3. Submits a watch request to monitor changes
+4. Maintains local collections of `acquired` and `waiting` entries
+5. Manages entries based on capacity and sequence order
+
+![](semaphore_diagram.svg)
+
+### Consistency Guarantees
+
+- Atomic verification of sequence numbers during insertion
+- Fair ordering based on sequence numbers
+- Automatic lease management to prevent stale entries
+- Real-time updates through meta-service watch API
+
+## Current Limitations
+
+- Capacity is set per client rather than globally (planned for future versions)
+- Different capacity values across clients may lead to starvation
+- Future versions will store capacity in the `meta` key for consistency
+
+## License
+
+This project is licensed under the Apache License 2.0 - see the LICENSE file for details.