Repository: cloudnative-pg/cloudnative-pg

Issue:
[Feature]: Startup and readiness probes for replicas
### Is there an existing issue already for this feature request/idea?

- [x] I have searched for an existing issue, and could not find anything. I believe this is a new feature request to be evaluated.

### What problem is this feature going to solve? Why should it be added?

The current implementation of the startup probe uses the same endpoint as the liveness probe, which is `/healthz`. This endpoint relies on the [`pg_isready` command](https://www.postgresql.org/docs/current/app-pg-isready.html). The liveness probe reports success when `pg_isready` returns either 0 (indicating that the system is ready for connections) or 1 (indicating that the system is starting up, such as during crash recovery).

The readiness probe begins after this point. However, in some cases, such as with replicas, this implementation may pose a limitation.

### Describe the solution you'd like

We would like to differentiate the startup probe to define multiple waiting stages before the readiness probe kicks in:

- `pg_isready`: pg_isready returns 0 (connection is possible)
- `streaming`: in case of a streaming connected replica, the WAL receiver is up and `lag` is within a desired bound


### Describe alternatives you've considered

The issue arose from a request by one of our customers at EDB to "tune" the instances participating in the synchronous replication quorum, with a focus on prioritising data durability. Following the shutdown of their only synchronous replica, the replica was automatically added to the `synchronous_standby_names` list as "ready" when it restarted. This caused write operations on the primary to halt until the replica became fully synchronous.

Our initial approach was to address this issue at the level of synchronous replication, pushing the problem to PostgreSQL's handling of replica readiness. However, after extensive discussions and scenario analysis between @leonardoce and me, we realised that modifying the readiness probe wasn't the optimal solution (better, wasn't the right place to solve it). Instead, the better approach would have been to focus on the startup probe, which operates exclusively during the pod's startup phase.

### Additional context

N/A

### Backport?

No

### Are you willing to actively contribute to this feature?

No

### Code of Conduct

- [x] I agree to follow this project's Code of Conduct

Hints from issue comments:

Solution patch:
diff --git a/.wordlist-en-custom.txt b/.wordlist-en-custom.txt
index a0f5490e..fe66ba48 100644
--- a/.wordlist-en-custom.txt
+++ b/.wordlist-en-custom.txt
@@ -343,7 +343,9 @@ PrimaryUpdateMethod
 PrimaryUpdateStrategy
 PriorityClass
 PriorityClassName
+ProbeStrategyType
 ProbeTerminationGracePeriod
+ProbeWithStrategy
 ProbesConfiguration
 ProjectedVolumeSource
 PublicationReclaimPolicy
@@ -442,6 +444,8 @@ SnapshotType
 Snapshotting
 Snyk
 Stackgres
+StartupProbe
+StartupStrategyType
 StatefulSets
 StorageClass
 StorageConfiguration
@@ -957,6 +961,7 @@ maxClientConnections
 maxParallel
 maxStandbyNamesFromCluster
 maxSyncReplicas
+maximumLag
 maxwait
 mcache
 md
@@ -1348,6 +1353,7 @@ tx
 ubi
 ui
 uid
+uint
 ul
 un
 uncordon
diff --git a/api/v1/cluster_funcs.go b/api/v1/cluster_funcs.go
index 3fd8e8aa..340ce9e7 100644
--- a/api/v1/cluster_funcs.go
+++ b/api/v1/cluster_funcs.go
@@ -1482,6 +1482,16 @@ func (p *Probe) ApplyInto(k8sProbe *corev1.Probe) {
 	}
 }
 
+// ApplyInto applies the content of the probe configuration in a Kubernetes
+// probe
+func (p *ProbeWithStrategy) ApplyInto(k8sProbe *corev1.Probe) {
+	if p == nil {
+		return
+	}
+
+	p.Probe.ApplyInto(k8sProbe)
+}
+
 // GetEnabledWALArchivePluginName returns the name of the enabled backup plugin or an empty string
 // if no backup plugin is enabled
 func (cluster *Cluster) GetEnabledWALArchivePluginName() string {
diff --git a/api/v1/cluster_types.go b/api/v1/cluster_types.go
index 97b9ed3b..70b5c2ed 100644
--- a/api/v1/cluster_types.go
+++ b/api/v1/cluster_types.go
@@ -487,15 +487,49 @@ type ClusterSpec struct {
 // to be injected in the PostgreSQL Pods
 type ProbesConfiguration struct {
 	// The startup probe configuration
-	Startup *Probe `json:"startup,omitempty"`
+	Startup *ProbeWithStrategy `json:"startup,omitempty"`
 
 	// The liveness probe configuration
 	Liveness *Probe `json:"liveness,omitempty"`
 
 	// The readiness probe configuration
-	Readiness *Probe `json:"readiness,omitempty"`
+	Readiness *ProbeWithStrategy `json:"readiness,omitempty"`
 }
 
+// ProbeWithStrategy is the configuration of the startup probe
+type ProbeWithStrategy struct {
+	// Probe is the standard probe configuration
+	Probe `json:",inline"`
+
+	// The probe strategy
+	// +kubebuilder:validation:Enum=pg_isready;streaming;query
+	// +optional
+	Type ProbeStrategyType `json:"type,omitempty"`
+
+	// Lag limit. Used only for `streaming` strategy
+	// +optional
+	MaximumLag *resource.Quantity `json:"maximumLag,omitempty"`
+}
+
+// ProbeStrategyType is the type of the strategy used to declare a PostgreSQL instance
+// ready
+type ProbeStrategyType string
+
+const (
+	// ProbeStrategyPgIsReady means that the pg_isready tool is used to determine
+	// whether PostgreSQL is started up
+	ProbeStrategyPgIsReady ProbeStrategyType = "pg_isready"
+
+	// ProbeStrategyStreaming means that pg_isready is positive and the replica is
+	// connected via streaming replication to the current primary and the lag is, if specified,
+	// within the limit.
+	ProbeStrategyStreaming ProbeStrategyType = "streaming"
+
+	// ProbeStrategyQuery means that the server is able to connect to the superuser database
+	// and able to execute a simple query like "-- ping"
+	ProbeStrategyQuery ProbeStrategyType = "query"
+)
+
 // Probe describes a health check to be performed against a container to determine whether it is
 // alive or ready to receive traffic.
 type Probe struct {
diff --git a/api/v1/zz_generated.deepcopy.go b/api/v1/zz_generated.deepcopy.go
index e6b7801f..a2d4e4c1 100644
--- a/api/v1/zz_generated.deepcopy.go
+++ b/api/v1/zz_generated.deepcopy.go
@@ -2202,12 +2202,33 @@ func (in *Probe) DeepCopy() *Probe {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ProbeWithStrategy) DeepCopyInto(out *ProbeWithStrategy) {
+	*out = *in
+	in.Probe.DeepCopyInto(&out.Probe)
+	if in.MaximumLag != nil {
+		in, out := &in.MaximumLag, &out.MaximumLag
+		x := (*in).DeepCopy()
+		*out = &x
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ProbeWithStrategy.
+func (in *ProbeWithStrategy) DeepCopy() *ProbeWithStrategy {
+	if in == nil {
+		return nil
+	}
+	out := new(ProbeWithStrategy)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ProbesConfiguration) DeepCopyInto(out *ProbesConfiguration) {
 	*out = *in
 	if in.Startup != nil {
 		in, out := &in.Startup, &out.Startup
-		*out = new(Probe)
+		*out = new(ProbeWithStrategy)
 		(*in).DeepCopyInto(*out)
 	}
 	if in.Liveness != nil {
@@ -2217,7 +2238,7 @@ func (in *ProbesConfiguration) DeepCopyInto(out *ProbesConfiguration) {
 	}
 	if in.Readiness != nil {
 		in, out := &in.Readiness, &out.Readiness
-		*out = new(Probe)
+		*out = new(ProbeWithStrategy)
 		(*in).DeepCopyInto(*out)
 	}
 }
diff --git a/config/crd/bases/postgresql.cnpg.io_clusters.yaml b/config/crd/bases/postgresql.cnpg.io_clusters.yaml
index ed7ece5b..7fcdabd3 100644
--- a/config/crd/bases/postgresql.cnpg.io_clusters.yaml
+++ b/config/crd/bases/postgresql.cnpg.io_clusters.yaml
@@ -4302,6 +4302,13 @@ spec:
                           More info: https://kubernetes.io/docs/concepts/workloads/pods/pod-lifecycle#container-probes
                         format: int32
                         type: integer
+                      maximumLag:
+                        anyOf:
+                        - type: integer
+                        - type: string
+                        description: Lag limit. Used only for `streaming` strategy
+                        pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                        x-kubernetes-int-or-string: true
                       periodSeconds:
                         description: |-
                           How often (in seconds) to perform the probe.
@@ -4335,6 +4342,13 @@ spec:
                           More info: https://kubernetes.io/docs/concepts/workloads/pods/pod-lifecycle#container-probes
                         format: int32
                         type: integer
+                      type:
+                        description: The probe strategy
+                        enum:
+                        - pg_isready
+                        - streaming
+                        - query
+                        type: string
                     type: object
                   startup:
                     description: The startup probe configuration
@@ -4351,6 +4365,13 @@ spec:
                           More info: https://kubernetes.io/docs/concepts/workloads/pods/pod-lifecycle#container-probes
                         format: int32
                         type: integer
+                      maximumLag:
+                        anyOf:
+                        - type: integer
+                        - type: string
+                        description: Lag limit. Used only for `streaming` strategy
+                        pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                        x-kubernetes-int-or-string: true
                       periodSeconds:
                         description: |-
                           How often (in seconds) to perform the probe.
@@ -4384,6 +4405,13 @@ spec:
                           More info: https://kubernetes.io/docs/concepts/workloads/pods/pod-lifecycle#container-probes
                         format: int32
                         type: integer
+                      type:
+                        description: The probe strategy
+                        enum:
+                        - pg_isready
+                        - streaming
+                        - query
+                        type: string
                     type: object
                 type: object
               projectedVolumeTemplate:
diff --git a/docs/src/cloudnative-pg.v1.md b/docs/src/cloudnative-pg.v1.md
index 74e43468..eb29b445 100644
--- a/docs/src/cloudnative-pg.v1.md
+++ b/docs/src/cloudnative-pg.v1.md
@@ -4216,6 +4216,8 @@ the primary server of the cluster as part of rolling updates</p>
 
 **Appears in:**
 
+- [ProbeWithStrategy](#postgresql-cnpg-io-v1-ProbeWithStrategy)
+
 - [ProbesConfiguration](#postgresql-cnpg-io-v1-ProbesConfiguration)
 
 
@@ -4286,6 +4288,59 @@ Minimum value is 1. spec.terminationGracePeriodSeconds is used if unset.</p>
 </tbody>
 </table>
 
+## ProbeStrategyType     {#postgresql-cnpg-io-v1-ProbeStrategyType}
+
+(Alias of `string`)
+
+**Appears in:**
+
+- [ProbeWithStrategy](#postgresql-cnpg-io-v1-ProbeWithStrategy)
+
+
+<p>ProbeStrategyType is the type of the strategy used to declare a PostgreSQL instance
+ready</p>
+
+
+
+
+## ProbeWithStrategy     {#postgresql-cnpg-io-v1-ProbeWithStrategy}
+
+
+**Appears in:**
+
+- [ProbesConfiguration](#postgresql-cnpg-io-v1-ProbesConfiguration)
+
+
+<p>ProbeWithStrategy is the configuration of the startup probe</p>
+
+
+<table class="table">
+<thead><tr><th width="30%">Field</th><th>Description</th></tr></thead>
+<tbody>
+<tr><td><code>Probe</code><br/>
+<a href="#postgresql-cnpg-io-v1-Probe"><i>Probe</i></a>
+</td>
+<td>(Members of <code>Probe</code> are embedded into this type.)
+   <p>Probe is the standard probe configuration</p>
+</td>
+</tr>
+<tr><td><code>type</code><br/>
+<a href="#postgresql-cnpg-io-v1-ProbeStrategyType"><i>ProbeStrategyType</i></a>
+</td>
+<td>
+   <p>The probe strategy</p>
+</td>
+</tr>
+<tr><td><code>maximumLag</code><br/>
+<a href="https://pkg.go.dev/k8s.io/apimachinery/pkg/api/resource#Quantity"><i>k8s.io/apimachinery/pkg/api/resource.Quantity</i></a>
+</td>
+<td>
+   <p>Lag limit. Used only for <code>streaming</code> strategy</p>
+</td>
+</tr>
+</tbody>
+</table>
+
 ## ProbesConfiguration     {#postgresql-cnpg-io-v1-ProbesConfiguration}
 
 
@@ -4302,7 +4357,7 @@ to be injected in the PostgreSQL Pods</p>
 <thead><tr><th width="30%">Field</th><th>Description</th></tr></thead>
 <tbody>
 <tr><td><code>startup</code> <B>[Required]</B><br/>
-<a href="#postgresql-cnpg-io-v1-Probe"><i>Probe</i></a>
+<a href="#postgresql-cnpg-io-v1-ProbeWithStrategy"><i>ProbeWithStrategy</i></a>
 </td>
 <td>
    <p>The startup probe configuration</p>
@@ -4316,7 +4371,7 @@ to be injected in the PostgreSQL Pods</p>
 </td>
 </tr>
 <tr><td><code>readiness</code> <B>[Required]</B><br/>
-<a href="#postgresql-cnpg-io-v1-Probe"><i>Probe</i></a>
+<a href="#postgresql-cnpg-io-v1-ProbeWithStrategy"><i>ProbeWithStrategy</i></a>
 </td>
 <td>
    <p>The readiness probe configuration</p>
diff --git a/docs/src/instance_manager.md b/docs/src/instance_manager.md
index 30dceab7..69032b99 100644
--- a/docs/src/instance_manager.md
+++ b/docs/src/instance_manager.md
@@ -15,23 +15,23 @@ main container, which in turn runs the PostgreSQL instance. During the lifetime
 of the Pod, the instance manager acts as a backend to handle the
 [startup, liveness and readiness probes](https://kubernetes.io/docs/concepts/workloads/pods/pod-lifecycle/#container-probes).
 
-## Startup, Liveness, and Readiness Probes
+## Startup Probe
 
-CloudNativePG leverages [PostgreSQL's `pg_isready`](https://www.postgresql.org/docs/current/app-pg-isready.html)
-to implement Kubernetes startup, liveness, and readiness probes.
+The startup probe ensures that a PostgreSQL instance, whether a primary or
+standby, has fully started.
 
-### Startup Probe
+!!! Info
+    By default, the startup probe uses
+    [`pg_isready`](https://www.postgresql.org/docs/current/app-pg-isready.html).
+    However, the behavior can be customized by specifying a different startup
+    strategy.
 
-The startup probe ensures that a PostgreSQL instance, whether a primary or
-standby, has fully started according to `pg_isready`.
 While the startup probe is running, the liveness and readiness probes remain
 disabled. Following Kubernetes standards, if the startup probe fails, the
 kubelet will terminate the container, which will then be restarted.
 
-The startup probe provided by CloudNativePG is configurable via the
-parameter `.spec.startDelay`, which specifies the maximum time, in seconds,
-allowed for the startup probe to succeed. At a minimum, the probe requires
-`pg_isready` to return `0` or `1`.
+The `.spec.startDelay` parameter specifies the maximum time, in seconds,
+allowed for the startup probe to succeed.
 
 By default, the `startDelay` is set to `3600` seconds. It is recommended to
 adjust this setting based on the time PostgreSQL needs to fully initialize in
@@ -63,7 +63,7 @@ section of your configuration.
 
 !!! Info
     For more details on probe configuration, refer to the
-    [probe API documentation](cloudnative-pg.v1.md#postgresql-cnpg-io-v1-Probe).
+    [probe API documentation](cloudnative-pg.v1.md#postgresql-cnpg-io-v1-ProbeWithStrategy).
 
 If you manually specify `.spec.probes.startup.failureThreshold`, it will
 override the default behavior and disable the automatic use of `startDelay`.
@@ -81,16 +81,63 @@ spec:
       failureThreshold: 10
 ```
 
-### Liveness Probe
+### Startup Probe Strategy
+
+In certain scenarios, you may need to customize the startup strategy for your
+PostgreSQL cluster. For example, you might delay marking a replica as started
+until it begins streaming from the primary or define a replication lag
+threshold that must be met before considering the replica ready.
+
+To accommodate these requirements, CloudNativePG extends the
+`.spec.probes.startup` stanza with two optional parameters:
+
+- `type`: specifies the criteria for considering the probe successful. Accepted
+  values, in increasing order of complexity/depth, include:
+
+    - `pg_isready`: marks the probe as successful when the `pg_isready` command
+      exits with `0`. This is the default for primary instances and replicas.
+    - `query`: marks the probe as successful when a basic query is executed on
+      the `postgres` database locally.
+    - `streaming`: marks the probe as successful when the replica begins
+      streaming from its source and meets the specified lag requirements (details
+      below).
+
+- `maximumLag`: defines the maximum acceptable replication lag, measured in bytes
+  (expressed as Kubernetes quantities). This parameter is only applicable when
+  `type` is set to `streaming`. If `maximumLag` is not specified, the replica is
+  considered successfully started as soon as it begins streaming.
+
+!!! Important
+    The `.spec.probes.startup.maximumLag` option is validated and enforced only
+    during the startup phase of the pod, meaning it applies exclusively when the
+    replica is starting.
+
+!!! Warning
+    Incorrect configuration of the `maximumLag` option can cause continuous
+    failures of the startup probe, leading to repeated replica restarts. Ensure
+    you understand how this option works and configure appropriate values for
+    `failureThreshold` and `periodSeconds` to give the replica enough time to
+    catch up with its source.
+
+The following example requires a replica to have a maximum lag of 16Mi from the
+source to be considered started:
+
+```yaml
+# <snip>
+probes:
+  startup:
+    type: streaming
+    maximumLag: 16Mi
+```
+
+## Liveness Probe
 
 The liveness probe begins after the startup probe successfully completes. Its
-primary role is to ensure the PostgreSQL instance—whether primary or standby—is
-operating correctly. This is achieved using the `pg_isready` utility. Both exit
-codes `0` (indicating the server is accepting connections) and `1` (indicating
-the server is rejecting connections, such as during startup or a smart
-shutdown) are treated as valid outcomes.
-Following Kubernetes standards, if the liveness probe fails, the
-kubelet will terminate the container, which will then be restarted.
+primary role is to ensure the PostgreSQL instance manager is operating
+correctly.
+
+Following Kubernetes standards, if the liveness probe fails, the kubelet will
+terminate the container, which will then be restarted.
 
 The amount of time before a Pod is classified as not alive is configurable via
 the `.spec.livenessProbeTimeout` parameter.
@@ -140,14 +187,21 @@ spec:
       failureThreshold: 10
 ```
 
-### Readiness Probe
+## Readiness Probe
+
+The readiness probe starts once the startup probe has successfully completed.
+Its primary purpose is to check whether the PostgreSQL instance is ready to
+accept traffic and serve requests at any point during the pod's lifecycle.
+
+!!! Info
+    By default, the readiness probe uses
+    [`pg_isready`](https://www.postgresql.org/docs/current/app-pg-isready.html).
+    However, the behavior can be customized by specifying a different readiness
+    strategy.
 
-The readiness probe begins once the startup probe has successfully completed.
-Its purpose is to check whether the PostgreSQL instance is ready to accept
-traffic and serve requests.
-For streaming replicas, it also requires that they have connected to the source
-at least once. Following Kubernetes standards, if the readiness probe fails,
-the pod will be marked unready and will not receive traffic from any services.
+Following Kubernetes standards, if the readiness probe fails, the pod will be
+marked unready and will not receive traffic from any services. An unready pod
+is also ineligible for promotion during automated failover scenarios.
 
 CloudNativePG uses the following default configuration for the readiness probe:
 
@@ -178,7 +232,53 @@ spec:
 
 !!! Info
     For more information on configuring probes, see the
-    [probe API](cloudnative-pg.v1.md#postgresql-cnpg-io-v1-Probe).
+    [probe API](cloudnative-pg.v1.md#postgresql-cnpg-io-v1-ProbeWithStrategy).
+
+### Readiness Probe Strategy
+
+In certain scenarios, you may need to customize the readiness strategy for your
+cluster. For example, you might delay marking a replica as ready until it
+begins streaming from the primary or define a maximum replication lag threshold
+before considering the replica ready.
+
+To accommodate these requirements, CloudNativePG extends the
+`.spec.probes.readiness` stanza with two optional parameters: `type` and
+`maximumLag`. Please refer to the [Startup Probe Strategy](#startup-probe-strategy)
+section for detailed information on these options.
+
+!!! Important
+    Unlike the startup probe, the `.spec.probes.readiness.maximumLag` option is
+    continuously monitored. A lagging replica may become unready if this setting is
+    not appropriately tuned.
+
+!!! Warning
+    Incorrect configuration of the `maximumLag` option can lead to repeated
+    readiness probe failures, causing serious consequences, such as:
+
+    - Exclusion of the replica from key operator features, such as promotion
+      during failover or participation in synchronous replication quorum.
+    - Disruptions in read/read-only services.
+    - In longer failover times scenarios, replicas might be declared unready,
+      leading to a cluster stall requiring manual intervention.
+
+!!! Recommendation
+    Use the `streaming` and `maximumLag` options with extreme caution. If
+    you're unfamiliar with PostgreSQL replication, rely on the default
+    strategy. Seek professional advice if unsure.
+
+The following example requires a replica to have a maximum lag of 64Mi from the
+source to be considered ready. It also provides approximately 300 seconds (30
+failures × 10 seconds) for the startup probe to succeed:
+
+```yaml
+# <snip>
+probes:
+  readiness:
+    type: streaming
+    maximumLag: 64Mi
+    failureThreshold: 30
+    periodSeconds: 10
+```
 
 ## Shutdown control
 
diff --git a/docs/src/release_notes/v1.26.md b/docs/src/release_notes/v1.26.md
index e25a2b46..f7749bb8 100644
--- a/docs/src/release_notes/v1.26.md
+++ b/docs/src/release_notes/v1.26.md
@@ -17,6 +17,10 @@ on the release branch in GitHub.
 
 ### Features:
 
+- **Improved Startup and Readiness Probes for Replicas**: Enhanced support for
+  Kubernetes startup and readiness probes in PostgreSQL instances, providing
+  greater control over replicas based on the streaming lag. (#6623)
+
 - **MAIN FEATURE #1**: short description
 - **MAIN FEATURE #2**: short description
 
diff --git a/docs/src/replication.md b/docs/src/replication.md
index fbe01b2d..30913336 100644
--- a/docs/src/replication.md
+++ b/docs/src/replication.md
@@ -394,7 +394,9 @@ number of standbys are available.
     Make sure you have a clear understanding of what *ready/available* means
     for a replica and set your expectations accordingly. By default, a replica is
     considered ready when it has successfully connected to the source at least
-    once.
+    once. However, CloudNativePG allows you to configure startup probes for
+    replicas. For more details, please refer to the
+    ["Advanced Startup Probe" section](instance_manager.md#advanced-startup-probe).
 
 This setting balances data safety with availability, enabling applications to
 continue writing during temporary standby unavailability—hence, it’s also known
diff --git a/docs/src/samples/cluster-example-with-probes.yaml b/docs/src/samples/cluster-example-with-probes.yaml
new file mode 100644
index 00000000..dc2c1bb9
--- /dev/null
+++ b/docs/src/samples/cluster-example-with-probes.yaml
@@ -0,0 +1,16 @@
+apiVersion: postgresql.cnpg.io/v1
+kind: Cluster
+metadata:
+  name: cluster-example
+spec:
+  instances: 3
+
+  probes:
+    startup:
+      type: streaming
+      maximumLag: 16Mi
+      failureThreshold: 30
+      periodSeconds: 10
+
+  storage:
+    size: 1Gi
diff --git a/internal/management/controller/instance_controller.go b/internal/management/controller/instance_controller.go
index c1386796..31d29b6e 100644
--- a/internal/management/controller/instance_controller.go
+++ b/internal/management/controller/instance_controller.go
@@ -186,7 +186,7 @@ func (r *InstanceReconciler) Reconcile(
 		return reconcile.Result{}, nil
 	}
 
-	if r.instance.IsServerHealthy() != nil {
+	if err := r.instance.IsReady(); err != nil {
 		contextLogger.Info("Instance is still down, will retry in 1 second")
 		return reconcile.Result{RequeueAfter: time.Second}, nil
 	}
diff --git a/internal/management/controller/roles/runnable.go b/internal/management/controller/roles/runnable.go
index 1d97e6bf..585dea7a 100644
--- a/internal/management/controller/roles/runnable.go
+++ b/internal/management/controller/roles/runnable.go
@@ -53,7 +53,7 @@ type instanceInterface interface {
 	GetSuperUserDB() (*sql.DB, error)
 	IsPrimary() (bool, error)
 	RoleSynchronizerChan() <-chan *apiv1.ManagedConfiguration
-	IsServerHealthy() error
+	IsReady() error
 	GetClusterName() string
 	GetNamespaceName() string
 }
@@ -134,7 +134,7 @@ func (sr *RoleSynchronizer) reconcile(ctx context.Context, config *apiv1.Managed
 	contextLog := log.FromContext(ctx).WithName("roles_reconciler")
 	contextLog.Debug("reconciling managed roles")
 
-	if sr.instance.IsServerHealthy() != nil {
+	if err := sr.instance.IsReady(); err != nil {
 		contextLog.Debug("database not ready, skipping roles reconciling")
 		return nil
 	}
diff --git a/internal/management/controller/tablespaces/manager.go b/internal/management/controller/tablespaces/manager.go
index 6ba7490a..0ea31706 100644
--- a/internal/management/controller/tablespaces/manager.go
+++ b/internal/management/controller/tablespaces/manager.go
@@ -35,6 +35,7 @@ type instanceInterface interface {
 	GetClusterName() string
 	GetSuperUserDB() (*sql.DB, error)
 	IsPrimary() (bool, error)
+	IsReady() error
 	CanCheckReadiness() bool
 }
 
diff --git a/internal/management/controller/tablespaces/reconciler.go b/internal/management/controller/tablespaces/reconciler.go
index 031ea0bf..c8f10295 100644
--- a/internal/management/controller/tablespaces/reconciler.go
+++ b/internal/management/controller/tablespaces/reconciler.go
@@ -29,7 +29,6 @@ import (
 
 	apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
 	"github.com/cloudnative-pg/cloudnative-pg/internal/management/controller/tablespaces/infrastructure"
-	"github.com/cloudnative-pg/cloudnative-pg/pkg/management/postgres/readiness"
 )
 
 // Reconcile is the main reconciliation loop for the instance
@@ -71,9 +70,10 @@ func (r *TablespaceReconciler) Reconcile(
 		return reconcile.Result{}, nil
 	}
 
-	checker := readiness.ForInstance(r.instance)
-	if checker.IsServerReady(ctx) != nil {
-		contextLogger.Debug("database not ready, skipping tablespace reconciling")
+	if err := r.instance.IsReady(); err != nil {
+		contextLogger.Debug(
+			"database not ready, skipping tablespace reconciling",
+			"err", err)
 		return reconcile.Result{RequeueAfter: time.Second}, nil
 	}
 
diff --git a/pkg/management/postgres/instance.go b/pkg/management/postgres/instance.go
index 362172bd..5bd3ae3c 100644
--- a/pkg/management/postgres/instance.go
+++ b/pkg/management/postgres/instance.go
@@ -225,6 +225,11 @@ func (instance *Instance) SetPostgreSQLAutoConfWritable(writeable bool) error {
 	return os.Chmod(autoConfFileName, mode)
 }
 
+// IsReady runs PgIsReady
+func (instance *Instance) IsReady() error {
+	return PgIsReady()
+}
+
 // IsFenced checks whether the instance is marked as fenced
 func (instance *Instance) IsFenced() bool {
 	return instance.fenced.Load()
diff --git a/pkg/management/postgres/probes.go b/pkg/management/postgres/probes.go
index 832d57a1..84144f16 100644
--- a/pkg/management/postgres/probes.go
+++ b/pkg/management/postgres/probes.go
@@ -35,20 +35,6 @@ import (
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/versions"
 )
 
-// IsServerHealthy check if the instance is healthy
-func (instance *Instance) IsServerHealthy() error {
-	err := PgIsReady()
-
-	// A healthy server can also be actively rejecting connections.
-	// That's not a problem: it's only the server starting up or shutting
-	// down.
-	if errors.Is(err, ErrPgRejectingConnection) {
-		return nil
-	}
-
-	return err
-}
-
 // GetStatus Extract the status of this PostgreSQL database
 func (instance *Instance) GetStatus() (result *postgres.PostgresqlStatus, err error) {
 	result = &postgres.PostgresqlStatus{
diff --git a/pkg/management/postgres/readiness/readiness.go b/pkg/management/postgres/readiness/readiness.go
deleted file mode 100644
index c3598538..00000000
--- a/pkg/management/postgres/readiness/readiness.go
+++ /dev/null
@@ -1,101 +0,0 @@
-/*
-Copyright The CloudNativePG Contributors
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package readiness
-
-import (
-	"context"
-	"database/sql"
-	"errors"
-)
-
-// ErrStreamingReplicaNotConnected is raised for streaming replicas that never connected to its primary
-var ErrStreamingReplicaNotConnected = errors.New("streaming replica was never connected to the primary node")
-
-// instanceInterface represents the required behavior for use in the readiness probe
-type instanceInterface interface {
-	CanCheckReadiness() bool
-	GetSuperUserDB() (*sql.DB, error)
-}
-
-// Data is the readiness checker structure
-type Data struct {
-	instance instanceInterface
-
-	streamingReplicaValidated bool
-}
-
-// ForInstance creates a readiness checker for a certain instance
-func ForInstance(instance instanceInterface) *Data {
-	return &Data{
-		instance: instance,
-	}
-}
-
-// IsServerReady check if the instance is healthy and can really accept connections
-func (data *Data) IsServerReady(ctx context.Context) error {
-	if !data.instance.CanCheckReadiness() {
-		return errors.New("instance is not ready yet")
-	}
-	superUserDB, err := data.instance.GetSuperUserDB()
-	if err != nil {
-		return err
-	}
-
-	// We now check if the database is ready to accept
-	// connections
-	if err := superUserDB.PingContext(ctx); err != nil {
-		return err
-	}
-
-	// If we already validated this streaming replica, everything
-	// is fine
-	if data.streamingReplicaValidated {
-		return nil
-	}
-
-	// If this is a streaming replica, meaning that
-	// primary_conninfo is not empty, we won't declare it ready
-	// unless it connected one time successfully to its primary.
-	//
-	// We check this because a streaming replica that was
-	// never connected to the primary could be incoherent,
-	// and we want users to notice this as soon as possible
-	row := superUserDB.QueryRowContext(
-		ctx,
-		`
-		SELECT
-			NOT pg_catalog.pg_is_in_recovery()
-			OR (SELECT coalesce(setting, '') = '' FROM pg_catalog.pg_settings WHERE name = 'primary_conninfo')
-			OR pg_catalog.pg_last_wal_replay_lsn() IS NOT NULL
-		`,
-	)
-	if err := row.Err(); err != nil {
-		return err
-	}
-
-	var status bool
-	if err := row.Scan(&status); err != nil {
-		return err
-	}
-
-	if !status {
-		return ErrStreamingReplicaNotConnected
-	}
-
-	data.streamingReplicaValidated = true
-	return nil
-}
diff --git a/pkg/management/postgres/webserver/probes/checker.go b/pkg/management/postgres/webserver/probes/checker.go
new file mode 100644
index 00000000..c85bba8b
--- /dev/null
+++ b/pkg/management/postgres/webserver/probes/checker.go
@@ -0,0 +1,152 @@
+/*
+Copyright The CloudNativePG Contributors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package probes
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+
+	"github.com/cloudnative-pg/machinery/pkg/log"
+	"k8s.io/utils/ptr"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/management/postgres"
+)
+
+// probeType is the type of the probe
+type probeType string
+
+const (
+	// probeTypeReadiness is the readiness probe
+	probeTypeReadiness probeType = "readiness"
+	// probeTypeStartup is the startup probe
+	probeTypeStartup probeType = "startup"
+)
+
+type runner interface {
+	// IsHealthy evaluates the status of PostgreSQL. If the probe is positive,
+	// it returns a nil error, otherwise the error status describes why
+	// the probe is failing
+	IsHealthy(ctx context.Context, instance *postgres.Instance) error
+}
+
+// Checker executes the probe and writes the response to the request
+type Checker interface {
+	IsHealthy(ctx context.Context, w http.ResponseWriter)
+}
+
+type executor struct {
+	cli       client.Client
+	instance  *postgres.Instance
+	probeType probeType
+}
+
+// NewReadinessChecker creates a new instance of the readiness probe checker
+func NewReadinessChecker(
+	cli client.Client,
+	instance *postgres.Instance,
+) Checker {
+	return &executor{
+		cli:       cli,
+		instance:  instance,
+		probeType: probeTypeReadiness,
+	}
+}
+
+// NewStartupChecker creates a new instance of the startup probe checker
+func NewStartupChecker(
+	cli client.Client,
+	instance *postgres.Instance,
+) Checker {
+	return &executor{
+		cli:       cli,
+		instance:  instance,
+		probeType: probeTypeStartup,
+	}
+}
+
+// IsHealthy executes the underlying probe logic and writes a response to the request accordingly to the result obtained
+func (e *executor) IsHealthy(
+	ctx context.Context,
+	w http.ResponseWriter,
+) {
+	contextLogger := log.FromContext(ctx)
+
+	var cluster apiv1.Cluster
+	if err := e.cli.Get(
+		ctx,
+		client.ObjectKey{Namespace: e.instance.GetNamespaceName(), Name: e.instance.GetClusterName()},
+		&cluster,
+	); err != nil {
+		contextLogger.Warning(
+			fmt.Sprintf("%s check failed, cannot check Cluster definition", e.probeType),
+			"err", err.Error(),
+		)
+		http.Error(
+			w,
+			fmt.Sprintf("%s check failed cannot get Cluster definition: %s", e.probeType, err.Error()),
+			http.StatusInternalServerError,
+		)
+		return
+	}
+
+	probeRunner := getProbeRunnerFromCluster(e.probeType, cluster)
+	if err := probeRunner.IsHealthy(ctx, e.instance); err != nil {
+		contextLogger.Warning(fmt.Sprintf("%s probe failing", e.probeType), "err", err.Error())
+		http.Error(
+			w,
+			fmt.Sprintf("%s check failed: %s", e.probeType, err.Error()),
+			http.StatusInternalServerError,
+		)
+		return
+	}
+
+	contextLogger.Trace(fmt.Sprintf("%s probe succeeding", e.probeType))
+	_, _ = fmt.Fprint(w, "OK")
+}
+
+func getProbeRunnerFromCluster(probeType probeType, cluster apiv1.Cluster) runner {
+	var probe *apiv1.ProbeWithStrategy
+	if cluster.Spec.Probes != nil {
+		switch probeType {
+		case probeTypeStartup:
+			probe = cluster.Spec.Probes.Startup
+
+		case probeTypeReadiness:
+			probe = cluster.Spec.Probes.Readiness
+		}
+	}
+
+	switch {
+	case probe == nil:
+		return pgIsReadyChecker{}
+	case probe.Type == apiv1.ProbeStrategyPgIsReady:
+		return pgIsReadyChecker{}
+	case probe.Type == apiv1.ProbeStrategyQuery:
+		return pgQueryChecker{}
+	case probe.Type == apiv1.ProbeStrategyStreaming:
+		result := pgStreamingChecker{}
+		if probe.MaximumLag != nil {
+			result.maximumLag = ptr.To(probe.MaximumLag.AsDec().UnscaledBig().Uint64())
+		}
+		return result
+	}
+
+	return pgIsReadyChecker{}
+}
diff --git a/pkg/management/postgres/readiness/doc.go b/pkg/management/postgres/webserver/probes/doc.go
similarity index 81%
rename from pkg/management/postgres/readiness/doc.go
rename to pkg/management/postgres/webserver/probes/doc.go
index b9af9c89..2350a833 100644
--- a/pkg/management/postgres/readiness/doc.go
+++ b/pkg/management/postgres/webserver/probes/doc.go
@@ -14,6 +14,6 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-// Package readiness contains the code needed to check
-// if PostgreSQL is ready to accept client connections.
-package readiness
+// Package probes contains the implementation of startup, liveness and
+// readiness probes
+package probes
diff --git a/pkg/management/postgres/webserver/probes/isready.go b/pkg/management/postgres/webserver/probes/isready.go
new file mode 100644
index 00000000..35c8deb3
--- /dev/null
+++ b/pkg/management/postgres/webserver/probes/isready.go
@@ -0,0 +1,31 @@
+/*
+Copyright The CloudNativePG Contributors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package probes
+
+import (
+	"context"
+
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/management/postgres"
+)
+
+// pgIsReadyChecker checks is PostgreSQL is ready
+type pgIsReadyChecker struct{}
+
+// IsHealthy implements the runner interface
+func (pgIsReadyChecker) IsHealthy(_ context.Context, instance *postgres.Instance) error {
+	return instance.IsReady()
+}
diff --git a/pkg/management/postgres/webserver/probes/query.go b/pkg/management/postgres/webserver/probes/query.go
new file mode 100644
index 00000000..f61a4331
--- /dev/null
+++ b/pkg/management/postgres/webserver/probes/query.go
@@ -0,0 +1,41 @@
+/*
+Copyright The CloudNativePG Contributors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package probes
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/management/postgres"
+)
+
+// pgQueryChecker checks if the PostgreSQL server can execute a simple query
+type pgQueryChecker struct{}
+
+// IsHealthy implements the runner interface
+func (c pgQueryChecker) IsHealthy(ctx context.Context, instance *postgres.Instance) error {
+	superUserDB, err := instance.GetSuperUserDB()
+	if err != nil {
+		return fmt.Errorf("while getting superuser connection pool: %w", err)
+	}
+
+	if err := superUserDB.PingContext(ctx); err != nil {
+		return fmt.Errorf("while pinging database: %w", err)
+	}
+
+	return nil
+}
diff --git a/pkg/management/postgres/webserver/probes/streaming.go b/pkg/management/postgres/webserver/probes/streaming.go
new file mode 100644
index 00000000..99bd5cd8
--- /dev/null
+++ b/pkg/management/postgres/webserver/probes/streaming.go
@@ -0,0 +1,123 @@
+/*
+Copyright The CloudNativePG Contributors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package probes
+
+import (
+	"context"
+	"fmt"
+	"math"
+
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/management/postgres"
+)
+
+// pgStreamingChecker checks if the replica is connected via streaming
+// replication and, optionally, if the lag is within the specified maximum
+type pgStreamingChecker struct {
+	maximumLag *uint64
+}
+
+// IsHealthy implements the runner interface
+func (c pgStreamingChecker) IsHealthy(ctx context.Context, instance *postgres.Instance) error {
+	superUserDB, err := instance.GetSuperUserDB()
+	if err != nil {
+		return fmt.Errorf("while getting superuser connection pool: %w", err)
+	}
+
+	var configuredLag uint64 = math.MaxUint64
+	if c.maximumLag != nil {
+		configuredLag = *c.maximumLag
+	}
+
+	// At this point, the instance is already running.
+	// The startup probe succeeds if the instance satisfies any of the following conditions:
+	// - It is a primary instance.
+	// - It is a log shipping replica (including a designated primary).
+	// - It is a streaming replica with replication lag below the specified threshold.
+	//   If no lag threshold is specified, the startup probe succeeds if the replica has successfully connected
+	//   to its source at least once.
+	row := superUserDB.QueryRowContext(
+		ctx,
+		`
+        WITH
+          lag AS (
+            SELECT
+              (latest_end_lsn - pg_last_wal_replay_lsn()) AS value,
+              latest_end_time
+            FROM pg_catalog.pg_stat_wal_receiver
+          )
+        SELECT
+          CASE
+            WHEN NOT pg_is_in_recovery()
+              THEN true
+            WHEN (SELECT coalesce(setting, '') = '' FROM pg_catalog.pg_settings WHERE name = 'primary_conninfo')
+              THEN true
+            WHEN (SELECT value FROM lag) < $1
+              THEN true
+            ELSE false
+          END AS ready_to_start,
+          COALESCE((SELECT value FROM lag), 0) AS lag,
+          COALESCE((SELECT latest_end_time FROM lag), '-infinity') AS latest_end_time
+		`,
+		configuredLag,
+	)
+	if err := row.Err(); err != nil {
+		return fmt.Errorf("streaming replication check failed: %w", err)
+	}
+
+	var status bool
+	var detectedLag uint64
+	var latestEndTime string
+	if err := row.Scan(&status, &detectedLag, &latestEndTime); err != nil {
+		return fmt.Errorf("streaming replication check failed (scan): %w", err)
+	}
+
+	if !status {
+		if detectedLag > configuredLag {
+			return &ReplicaLaggingError{
+				DetectedLag:   detectedLag,
+				ConfiguredLag: configuredLag,
+				LatestEndTime: latestEndTime,
+			}
+		}
+		return fmt.Errorf("replica not connected via streaming replication")
+	}
+
+	return nil
+}
+
+// ReplicaLaggingError is raised when a replica is lagging more
+// than the configured cap
+type ReplicaLaggingError struct {
+	// DetectedLag is the lag that was detected
+	DetectedLag uint64
+
+	// ConfiguredLag is the lag as configured in the probe
+	ConfiguredLag uint64
+
+	// LatestEndTime is the time of last write-ahead log location reported to
+	// origin WAL sender
+	LatestEndTime string
+}
+
+func (e *ReplicaLaggingError) Error() string {
+	return fmt.Sprintf(
+		"streaming replica lagging; detectedLag=%v configuredLag=%v latestEndTime=%s",
+		e.DetectedLag,
+		e.ConfiguredLag,
+		e.LatestEndTime,
+	)
+}
diff --git a/pkg/management/postgres/webserver/remote.go b/pkg/management/postgres/webserver/remote.go
index a3c43eb5..4aeb7da9 100644
--- a/pkg/management/postgres/webserver/remote.go
+++ b/pkg/management/postgres/webserver/remote.go
@@ -42,7 +42,7 @@ import (
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/management"
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/management/postgres"
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/management/postgres/constants"
-	"github.com/cloudnative-pg/cloudnative-pg/pkg/management/postgres/readiness"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/management/postgres/webserver/probes"
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/management/upgrade"
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/management/url"
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/utils"
@@ -62,7 +62,6 @@ type remoteWebserverEndpoints struct {
 	typedClient          client.Client
 	instance             *postgres.Instance
 	currentBackup        *backupConnection
-	readinessChecker     *readiness.Data
 	ongoingBackupRequest sync.Mutex
 }
 
@@ -95,15 +94,15 @@ func NewRemoteWebServer(
 	}
 
 	endpoints := remoteWebserverEndpoints{
-		typedClient:      typedClient,
-		instance:         instance,
-		readinessChecker: readiness.ForInstance(instance),
+		typedClient: typedClient,
+		instance:    instance,
 	}
 
 	serveMux := http.NewServeMux()
 	serveMux.HandleFunc(url.PathPgModeBackup, endpoints.backup)
 	serveMux.HandleFunc(url.PathHealth, endpoints.isServerHealthy)
 	serveMux.HandleFunc(url.PathReady, endpoints.isServerReady)
+	serveMux.HandleFunc(url.PathStartup, endpoints.isServerStartedUp)
 	serveMux.HandleFunc(url.PathPgStatus, endpoints.pgStatus)
 	serveMux.HandleFunc(url.PathPgArchivePartial, endpoints.pgArchivePartial)
 	serveMux.HandleFunc(url.PathPGControlData, endpoints.pgControlData)
@@ -206,37 +205,33 @@ func (ws *remoteWebserverEndpoints) cleanupStaleCollections(ctx context.Context)
 	}
 }
 
-func (ws *remoteWebserverEndpoints) isServerHealthy(w http.ResponseWriter, _ *http.Request) {
-	// If `pg_rewind` is running the Pod is starting up.
+// isServerStartedUp evaluates the liveness probe
+func (ws *remoteWebserverEndpoints) isServerStartedUp(w http.ResponseWriter, req *http.Request) {
+	// If `pg_rewind` is running, it means that the Pod is starting up.
 	// We need to report it healthy to avoid being killed by the kubelet.
-	// Same goes for instances with fencing on.
 	if ws.instance.PgRewindIsRunning || ws.instance.MightBeUnavailable() {
-		log.Trace("Liveness probe skipped")
+		log.Trace("Startup probe skipped")
 		_, _ = fmt.Fprint(w, "Skipped")
 		return
 	}
 
-	err := ws.instance.IsServerHealthy()
-	if err != nil {
-		log.Debug("Liveness probe failing", "err", err.Error())
-		http.Error(w, err.Error(), http.StatusInternalServerError)
-		return
-	}
+	checker := probes.NewStartupChecker(ws.typedClient, ws.instance)
+	checker.IsHealthy(req.Context(), w)
+}
 
-	log.Trace("Liveness probe succeeding")
+func (ws *remoteWebserverEndpoints) isServerHealthy(w http.ResponseWriter, _ *http.Request) {
 	_, _ = fmt.Fprint(w, "OK")
 }
 
 // This is the readiness probe
-func (ws *remoteWebserverEndpoints) isServerReady(w http.ResponseWriter, r *http.Request) {
-	if err := ws.readinessChecker.IsServerReady(r.Context()); err != nil {
-		log.Debug("Readiness probe failing", "err", err.Error())
-		http.Error(w, err.Error(), http.StatusInternalServerError)
+func (ws *remoteWebserverEndpoints) isServerReady(w http.ResponseWriter, req *http.Request) {
+	if !ws.instance.CanCheckReadiness() {
+		http.Error(w, "instance is not ready yet", http.StatusInternalServerError)
 		return
 	}
 
-	log.Trace("Readiness probe succeeding")
-	_, _ = fmt.Fprint(w, "OK")
+	checker := probes.NewReadinessChecker(ws.typedClient, ws.instance)
+	checker.IsHealthy(req.Context(), w)
 }
 
 // This probe is for the instance status, including replication
diff --git a/pkg/management/url/url.go b/pkg/management/url/url.go
index 31e3b4ce..d18d94cc 100644
--- a/pkg/management/url/url.go
+++ b/pkg/management/url/url.go
@@ -34,9 +34,12 @@ const (
 	// PathHealth is the URL path for Health State
 	PathHealth string = "/healthz"
 
-	// PathReady is the URL oath for Ready State
+	// PathReady is the URL path for Ready State
 	PathReady string = "/readyz"
 
+	// PathStartup is the URL path for the Startup probe
+	PathStartup string = "/startupz"
+
 	// PathPGControlData is the URL path for PostgreSQL pg_controldata output
 	PathPGControlData string = "/pg/controldata"
 
diff --git a/pkg/specs/pods.go b/pkg/specs/pods.go
index 41482b5d..d059818f 100644
--- a/pkg/specs/pods.go
+++ b/pkg/specs/pods.go
@@ -206,7 +206,7 @@ func createPostgresContainers(cluster apiv1.Cluster, envConfig EnvConfig, enable
 				TimeoutSeconds: 5,
 				ProbeHandler: corev1.ProbeHandler{
 					HTTPGet: &corev1.HTTPGetAction{
-						Path: url.PathHealth,
+						Path: url.PathStartup,
 						Port: intstr.FromInt32(url.StatusPort),
 					},
 				},



Test patch:
diff --git a/internal/management/controller/tablespaces/controller_test.go b/internal/management/controller/tablespaces/controller_test.go
index 5d62d6da..000bd19a 100644
--- a/internal/management/controller/tablespaces/controller_test.go
+++ b/internal/management/controller/tablespaces/controller_test.go
@@ -71,6 +71,10 @@ func (f fakeInstance) IsPrimary() (bool, error) {
 	return true, nil
 }
 
+func (f fakeInstance) IsReady() error {
+	return nil
+}
+
 const (
 	expectedListStmt = `
 	SELECT
@@ -84,13 +88,6 @@ const (
 		"LOCATION '%s'"
 
 	expectedUpdateStmt = "ALTER TABLESPACE \"%s\" OWNER TO \"%s\""
-
-	expectedReadinessCheck = `
-		SELECT
-			NOT pg_catalog.pg_is_in_recovery()
-			OR (SELECT coalesce(setting, '') = '' FROM pg_catalog.pg_settings WHERE name = 'primary_conninfo')
-			OR pg_catalog.pg_last_wal_replay_lsn() IS NOT NULL
-		`
 )
 
 func getCluster(ctx context.Context, c client.Client, cluster *apiv1.Cluster) (*apiv1.Cluster, error) {
@@ -150,11 +147,6 @@ func assertTablespaceReconciled(ctx context.Context, tt tablespaceTest) {
 		storageManager: tt.storageManager,
 	}
 
-	// these bits happen because the reconciler checks for instance readiness
-	dbMock.ExpectPing()
-	expectedReadiness := sqlmock.NewRows([]string{""}).AddRow("t")
-	dbMock.ExpectQuery(expectedReadinessCheck).WillReturnRows(expectedReadiness)
-
 	tt.postgresExpectations(dbMock)
 
 	results, err := tablespaceReconciler.Reconcile(ctx, reconcile.Request{})
diff --git a/tests/e2e/fixtures/pg_data_corruption/cluster-pg-data-corruption-no-slots.yaml.template b/tests/e2e/fixtures/pg_data_corruption/cluster-pg-data-corruption-no-slots.yaml.template
index c38a36ff..d42e6076 100644
--- a/tests/e2e/fixtures/pg_data_corruption/cluster-pg-data-corruption-no-slots.yaml.template
+++ b/tests/e2e/fixtures/pg_data_corruption/cluster-pg-data-corruption-no-slots.yaml.template
@@ -27,6 +27,10 @@ spec:
       enabled: false
       slotPrefix: _cnpg_
 
+  probes:
+    readiness:
+      type: query
+
   # Persistent storage configuration
   storage:
     storageClass: ${E2E_DEFAULT_STORAGE_CLASS}
diff --git a/tests/e2e/fixtures/pg_data_corruption/cluster-pg-data-corruption-roles.yaml.template b/tests/e2e/fixtures/pg_data_corruption/cluster-pg-data-corruption-roles.yaml.template
index 29c869a3..d2cffe85 100644
--- a/tests/e2e/fixtures/pg_data_corruption/cluster-pg-data-corruption-roles.yaml.template
+++ b/tests/e2e/fixtures/pg_data_corruption/cluster-pg-data-corruption-roles.yaml.template
@@ -36,6 +36,10 @@ spec:
       database: app
       owner: app
 
+  probes:
+    readiness:
+      type: query
+
   # Persistent storage configuration
   storage:
     storageClass: ${E2E_DEFAULT_STORAGE_CLASS}
diff --git a/tests/e2e/fixtures/pg_data_corruption/cluster-pg-data-corruption.yaml.template b/tests/e2e/fixtures/pg_data_corruption/cluster-pg-data-corruption.yaml.template
index f46f729c..9c419300 100644
--- a/tests/e2e/fixtures/pg_data_corruption/cluster-pg-data-corruption.yaml.template
+++ b/tests/e2e/fixtures/pg_data_corruption/cluster-pg-data-corruption.yaml.template
@@ -21,6 +21,10 @@ spec:
       database: app
       owner: app
 
+  probes:
+    readiness:
+      type: query
+
   # Persistent storage configuration
   storage:
     storageClass: ${E2E_DEFAULT_STORAGE_CLASS}
diff --git a/tests/e2e/fixtures/sync_replicas/readiness-probe-lag-control.yaml.template b/tests/e2e/fixtures/sync_replicas/readiness-probe-lag-control.yaml.template
new file mode 100644
index 00000000..af46bd51
--- /dev/null
+++ b/tests/e2e/fixtures/sync_replicas/readiness-probe-lag-control.yaml.template
@@ -0,0 +1,31 @@
+apiVersion: postgresql.cnpg.io/v1
+kind: Cluster
+metadata:
+  name: cluster-readiness-probe-replica-lag-control
+spec:
+  instances: 3
+
+  postgresql:
+    synchronous:
+      method: any
+      number: 2
+      dataDurability: preferred
+    parameters:
+      log_checkpoints: "on"
+      log_lock_waits: "on"
+      log_min_duration_statement: '1000'
+      log_statement: 'ddl'
+      log_temp_files: '1024'
+      log_autovacuum_min_duration: '1s'
+      log_replication_commands: 'on'
+
+  probes:
+    readiness:
+      type: streaming
+      maximumLag: 16Mi
+      failureThreshold: 30
+      periodSeconds: 1
+
+  storage:
+    storageClass: ${E2E_DEFAULT_STORAGE_CLASS}
+    size: 1G
diff --git a/tests/e2e/fixtures/sync_replicas/startup-probe-lag-control.yaml.template b/tests/e2e/fixtures/sync_replicas/startup-probe-lag-control.yaml.template
new file mode 100644
index 00000000..adbefa38
--- /dev/null
+++ b/tests/e2e/fixtures/sync_replicas/startup-probe-lag-control.yaml.template
@@ -0,0 +1,31 @@
+apiVersion: postgresql.cnpg.io/v1
+kind: Cluster
+metadata:
+  name: cluster-startup-probe-replica-lag-control
+spec:
+  instances: 3
+
+  postgresql:
+    synchronous:
+      method: any
+      number: 2
+      dataDurability: preferred
+    parameters:
+      log_checkpoints: "on"
+      log_lock_waits: "on"
+      log_min_duration_statement: '1000'
+      log_statement: 'ddl'
+      log_temp_files: '1024'
+      log_autovacuum_min_duration: '1s'
+      log_replication_commands: 'on'
+
+  probes:
+    startup:
+      type: streaming
+      maximumLag: 16Mi
+      failureThreshold: 30
+      periodSeconds: 1
+
+  storage:
+    storageClass: ${E2E_DEFAULT_STORAGE_CLASS}
+    size: 1G
diff --git a/tests/e2e/probes_test.go b/tests/e2e/probes_test.go
index c3858d92..8bf817c6 100644
--- a/tests/e2e/probes_test.go
+++ b/tests/e2e/probes_test.go
@@ -56,9 +56,13 @@ var _ = Describe("Probes configuration tests", Label(tests.LabelBasic), func() {
 			TimeoutSeconds:      8,
 		}
 		probesConfiguration := apiv1.ProbesConfiguration{
-			Startup:   probeConfiguration.DeepCopy(),
-			Liveness:  probeConfiguration.DeepCopy(),
-			Readiness: probeConfiguration.DeepCopy(),
+			Startup: &apiv1.ProbeWithStrategy{
+				Probe: probeConfiguration,
+			},
+			Liveness: probeConfiguration.DeepCopy(),
+			Readiness: &apiv1.ProbeWithStrategy{
+				Probe: probeConfiguration,
+			},
 		}
 
 		assertProbeCoherentWithConfiguration := func(probe *corev1.Probe) {
@@ -70,7 +74,7 @@ var _ = Describe("Probes configuration tests", Label(tests.LabelBasic), func() {
 		assertProbesCoherentWithConfiguration := func(container *corev1.Container) {
 			assertProbeCoherentWithConfiguration(container.LivenessProbe)
 			assertProbeCoherentWithConfiguration(container.ReadinessProbe)
-			assertProbeCoherentWithConfiguration(container.LivenessProbe)
+			assertProbeCoherentWithConfiguration(container.StartupProbe)
 		}
 
 		var defaultReadinessProbe *corev1.Probe
diff --git a/tests/e2e/syncreplicas_test.go b/tests/e2e/syncreplicas_test.go
index 9731ae3f..80982e6d 100644
--- a/tests/e2e/syncreplicas_test.go
+++ b/tests/e2e/syncreplicas_test.go
@@ -22,14 +22,18 @@ import (
 	"strings"
 	"time"
 
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/utils/ptr"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/specs"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/utils"
 	"github.com/cloudnative-pg/cloudnative-pg/tests"
 	"github.com/cloudnative-pg/cloudnative-pg/tests/utils/clusterutils"
 	"github.com/cloudnative-pg/cloudnative-pg/tests/utils/exec"
 	"github.com/cloudnative-pg/cloudnative-pg/tests/utils/fencing"
+	"github.com/cloudnative-pg/cloudnative-pg/tests/utils/logs"
 	"github.com/cloudnative-pg/cloudnative-pg/tests/utils/run"
 	"github.com/cloudnative-pg/cloudnative-pg/tests/utils/yaml"
 
@@ -83,7 +87,71 @@ var _ = Describe("Synchronous Replicas", Label(tests.LabelReplication), func() {
 			Expect(err).ShouldNot(HaveOccurred())
 
 			return strings.Trim(out, "\n")
-		}, 30).Should(ContainSubstring(element))
+		}, 60).Should(ContainSubstring(element))
+	}
+
+	assertProbeRespectsReplicaLag := func(namespace, replicaName, probeType string) {
+		By(fmt.Sprintf(
+			"checking that %s probe of replica %s is waiting for lag to decrease before marking the pod ready",
+			probeType, replicaName), func() {
+			timeout := 2 * time.Minute
+
+			// This "Eventually" block is needed because we may grab only a portion
+			// of the replica logs, and the "ParseJSONLogs" function may fail on the latest
+			// log record when this happens
+			Eventually(func(g Gomega) {
+				data, err := logs.ParseJSONLogs(env.Ctx, env.Interface, namespace, replicaName)
+				g.Expect(err).ToNot(HaveOccurred())
+
+				recordWasFound := false
+				for _, record := range data {
+					err, ok := record["err"].(string)
+					if !ok {
+						continue
+					}
+					msg, ok := record["msg"].(string)
+					if !ok {
+						continue
+					}
+
+					if msg == fmt.Sprintf("%s probe failing", probeType) &&
+						strings.Contains(err, "streaming replica lagging") {
+						recordWasFound = true
+						break
+					}
+				}
+
+				g.Expect(recordWasFound).To(
+					BeTrue(),
+					fmt.Sprintf("The %s probe is preventing the replica from being marked ready", probeType),
+				)
+			}, timeout).Should(Succeed())
+		})
+	}
+
+	generateDataLoad := func(namespace, clusterName string) {
+		By("adding data to the primary", func() {
+			commandTimeout := time.Second * 600
+			primary, err := clusterutils.GetPrimary(env.Ctx, env.Client, namespace, clusterName)
+			Expect(err).ToNot(HaveOccurred())
+
+			// This will generate 1Gi of data in the primary node and, since the replica we fenced
+			// is not aligned, will generate lag.
+			_, _, err = exec.Command(
+				env.Ctx, env.Interface, env.RestClientConfig,
+				*primary, specs.PostgresContainerName, &commandTimeout,
+				"psql",
+				"-U",
+				"postgres",
+				"-c",
+				"create table numbers (i integer); "+
+					"insert into numbers (select generate_series(1,1000000)); "+
+					"insert into numbers (select * from numbers); "+
+					"insert into numbers (select * from numbers); "+
+					"insert into numbers (select * from numbers); ",
+			)
+			Expect(err).ToNot(HaveOccurred())
+		})
 	}
 
 	Context("Legacy synchronous replication", func() {
@@ -179,9 +247,8 @@ var _ = Describe("Synchronous Replicas", Label(tests.LabelReplication), func() {
 	})
 
 	Context("Synchronous replication", func() {
-		var namespace string
-
 		It("can manage quorum/priority based synchronous replication", func() {
+			var namespace string
 			const (
 				namespacePrefix = "sync-replicas-e2e"
 				sampleFile      = fixturesDir + "/sync_replicas/cluster-sync-replica.yaml.template"
@@ -240,6 +307,7 @@ var _ = Describe("Synchronous Replicas", Label(tests.LabelReplication), func() {
 
 		Context("data durability is preferred", func() {
 			It("will decrease the number of sync replicas to the number of available replicas", func() {
+				var namespace string
 				const (
 					namespacePrefix = "sync-replicas-preferred"
 					sampleFile      = fixturesDir + "/sync_replicas/preferred.yaml.template"
@@ -279,7 +347,7 @@ var _ = Describe("Synchronous Replicas", Label(tests.LabelReplication), func() {
 						return strings.Trim(stdout, "\n")
 					}, 160).Should(BeEmpty())
 				})
-				By("unfenicing the replicas and verifying we have 2 quorum-based replicas", func() {
+				By("unfencing the replicas and verifying we have 2 quorum-based replicas", func() {
 					Expect(fencing.Off(env.Ctx, env.Client, fmt.Sprintf("%v-3", clusterName),
 						namespace, clusterName, fencing.UsingAnnotation)).Should(Succeed())
 					Expect(fencing.Off(env.Ctx, env.Client, fmt.Sprintf("%v-2", clusterName),
@@ -289,5 +357,138 @@ var _ = Describe("Synchronous Replicas", Label(tests.LabelReplication), func() {
 				})
 			})
 		})
+
+		Context("Lag-control in startup & readiness probes", func() {
+			var (
+				namespace         string
+				namespacePrefix   string
+				sampleFile        string
+				clusterName       string
+				fencedReplicaName string
+				err               error
+			)
+
+			setupClusterWithLaggingReplica := func() {
+				clusterName, err = yaml.GetResourceNameFromYAML(env.Scheme, sampleFile)
+				Expect(err).ToNot(HaveOccurred())
+
+				namespace, err = env.CreateUniqueTestNamespace(env.Ctx, env.Client, namespacePrefix)
+				Expect(err).ToNot(HaveOccurred())
+				AssertCreateCluster(namespace, clusterName, sampleFile, env)
+
+				// Set our target fencedReplica
+				fencedReplicaName = fmt.Sprintf("%s-2", clusterName)
+
+				By("verifying we have 2 quorum-based replicas", func() {
+					getSyncReplicationCount(namespace, clusterName, "quorum", 2)
+					compareSynchronousStandbyNames(namespace, clusterName, "ANY 2")
+				})
+
+				By("fencing a replica and verifying we have only 1 quorum-based replica", func() {
+					Expect(fencing.On(env.Ctx, env.Client, fencedReplicaName,
+						namespace, clusterName, fencing.UsingAnnotation)).Should(Succeed())
+					getSyncReplicationCount(namespace, clusterName, "quorum", 1)
+					compareSynchronousStandbyNames(namespace, clusterName, "ANY 1")
+				})
+
+				By("waiting for the fenced pod to be not ready", func() {
+					Eventually(func(g Gomega) bool {
+						var pod corev1.Pod
+						err := env.Client.Get(env.Ctx, client.ObjectKey{
+							Namespace: namespace,
+							Name:      fencedReplicaName,
+						}, &pod)
+						g.Expect(err).ToNot(HaveOccurred())
+
+						return utils.IsPodReady(pod)
+					}, 160).Should(BeFalse())
+				})
+
+				generateDataLoad(namespace, clusterName)
+			}
+
+			It("lag control in startup probe will delay the readiness of replicas", func() {
+				namespacePrefix = "startup-probe-lag"
+				sampleFile = fixturesDir + "/sync_replicas/startup-probe-lag-control.yaml.template"
+
+				setupClusterWithLaggingReplica()
+
+				By("stopping the reconciliation loop on the cluster", func() {
+					// This is needed to avoid the operator to recreate the new Pod when we'll
+					// delete it.
+					// We want the Pod to start without being fenced to engage the lag checking
+					// startup probe
+					cluster, err := clusterutils.Get(env.Ctx, env.Client, namespace, clusterName)
+					Expect(err).ToNot(HaveOccurred())
+
+					origCluster := cluster.DeepCopy()
+					if cluster.Annotations == nil {
+						cluster.Annotations = make(map[string]string)
+					}
+					cluster.Annotations[utils.ReconciliationLoopAnnotationName] = "disabled"
+
+					err = env.Client.Patch(env.Ctx, cluster, client.MergeFrom(origCluster))
+					Expect(err).ToNot(HaveOccurred())
+				})
+
+				By("deleting the test replica and disabling fencing", func() {
+					var pod corev1.Pod
+					err := env.Client.Get(env.Ctx, client.ObjectKey{
+						Namespace: namespace,
+						Name:      fencedReplicaName,
+					}, &pod)
+					Expect(err).ToNot(HaveOccurred())
+
+					err = env.Client.Delete(env.Ctx, &pod)
+					Expect(err).ToNot(HaveOccurred())
+
+					Expect(fencing.Off(env.Ctx, env.Client, fmt.Sprintf("%v-2", clusterName),
+						namespace, clusterName, fencing.UsingAnnotation)).Should(Succeed())
+				})
+
+				By("enabling the reconciliation loops on the cluster", func() {
+					cluster, err := clusterutils.Get(env.Ctx, env.Client, namespace, clusterName)
+					Expect(err).ToNot(HaveOccurred())
+
+					origCluster := cluster.DeepCopy()
+					if cluster.Annotations == nil {
+						cluster.Annotations = make(map[string]string)
+					}
+					delete(cluster.Annotations, utils.ReconciliationLoopAnnotationName)
+
+					err = env.Client.Patch(env.Ctx, cluster, client.MergeFrom(origCluster))
+					Expect(err).ToNot(HaveOccurred())
+				})
+
+				By("waiting for the replica to be back again and ready", func() {
+					Eventually(func(g Gomega) bool {
+						var pod corev1.Pod
+						err := env.Client.Get(env.Ctx, client.ObjectKey{
+							Namespace: namespace,
+							Name:      fencedReplicaName,
+						}, &pod)
+						g.Expect(err).ToNot(HaveOccurred())
+
+						return utils.IsPodReady(pod)
+					}, 160).Should(BeTrue())
+				})
+
+				assertProbeRespectsReplicaLag(namespace, fencedReplicaName, "startup")
+			})
+
+			It("lag control in readiness probe will delay the readiness of replicas", func() {
+				namespacePrefix = "readiness-probe-lag"
+				sampleFile = fixturesDir + "/sync_replicas/readiness-probe-lag-control.yaml.template"
+
+				setupClusterWithLaggingReplica()
+
+				By("disabling fencing", func() {
+					Expect(fencing.Off(env.Ctx, env.Client, fmt.Sprintf("%v-2", clusterName),
+						namespace, clusterName, fencing.UsingAnnotation)).Should(Succeed())
+				})
+
+				assertProbeRespectsReplicaLag(namespace, fencedReplicaName, "readiness")
+			})
+		})
 	})
 })