Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * wait.c
4 : : * Implements WAIT, which allows waiting for events such as
5 : : * time passing or LSN having been replayed, flushed, or written.
6 : : *
7 : : * Portions Copyright (c) 2025-2026, PostgreSQL Global Development Group
8 : : *
9 : : * IDENTIFICATION
10 : : * src/backend/commands/wait.c
11 : : *
12 : : *-------------------------------------------------------------------------
13 : : */
14 : : #include "postgres.h"
15 : :
16 : : #include "access/xact.h"
17 : : #include "access/xlog.h"
18 : : #include "access/xlogrecovery.h"
19 : : #include "access/xlogwait.h"
20 : : #include "catalog/pg_type_d.h"
21 : : #include "commands/defrem.h"
22 : : #include "commands/wait.h"
23 : : #include "executor/executor.h"
24 : : #include "parser/parse_node.h"
25 : : #include "storage/lmgr.h"
26 : : #include "storage/lock.h"
27 : : #include "storage/proc.h"
28 : : #include "utils/builtins.h"
29 : : #include "utils/guc.h"
30 : : #include "utils/pg_lsn.h"
31 : : #include "utils/snapmgr.h"
32 : :
33 : :
34 : : void
35 : 290 : ExecWaitStmt(ParseState *pstate, WaitStmt *stmt, bool isTopLevel,
36 : : DestReceiver *dest)
37 : : {
38 : : XLogRecPtr lsn;
39 : 290 : int timeout = 0;
40 : : WaitLSNResult waitLSNResult;
41 : 290 : WaitLSNType lsnType = WAIT_LSN_TYPE_STANDBY_REPLAY; /* default */
42 : 290 : bool throw = true;
43 : : TupleDesc tupdesc;
44 : : TupOutputState *tstate;
45 : 290 : const char *result = "<unset>";
46 : 290 : bool timeout_specified = false;
47 : 290 : bool no_throw_specified = false;
48 : 290 : bool mode_specified = false;
49 : :
50 : : /*
51 : : * WAIT must not be run as a non-top-level statement (e.g., inside a
52 : : * function, procedure, or DO block). Forbid this case upfront.
53 : : */
54 [ + + ]: 290 : if (!isTopLevel)
55 [ + - ]: 3 : ereport(ERROR,
56 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
57 : : errmsg("%s can only be executed as a top-level statement",
58 : : "WAIT"),
59 : : errdetail("WAIT cannot be used within a function, procedure, or DO block.")));
60 : :
61 : : /* Parse and validate the mandatory LSN */
62 : 287 : lsn = DatumGetLSN(DirectFunctionCall1(pg_lsn_in,
63 : : CStringGetDatum(stmt->lsn_literal)));
64 : :
65 [ + + + + : 1303 : foreach_node(DefElem, defel, stmt->options)
+ + ]
66 : : {
67 [ + + ]: 749 : if (strcmp(defel->defname, "mode") == 0)
68 : : {
69 : : char *mode_str;
70 : :
71 [ + + ]: 258 : if (mode_specified)
72 : 1 : errorConflictingDefElem(defel, pstate);
73 : 257 : mode_specified = true;
74 : :
75 : 257 : mode_str = defGetString(defel);
76 : :
77 [ + + ]: 257 : if (pg_strcasecmp(mode_str, "standby_replay") == 0)
78 : 197 : lsnType = WAIT_LSN_TYPE_STANDBY_REPLAY;
79 [ + + ]: 60 : else if (pg_strcasecmp(mode_str, "standby_write") == 0)
80 : 29 : lsnType = WAIT_LSN_TYPE_STANDBY_WRITE;
81 [ + + ]: 31 : else if (pg_strcasecmp(mode_str, "standby_flush") == 0)
82 : 16 : lsnType = WAIT_LSN_TYPE_STANDBY_FLUSH;
83 [ + + ]: 15 : else if (pg_strcasecmp(mode_str, "primary_flush") == 0)
84 : 14 : lsnType = WAIT_LSN_TYPE_PRIMARY_FLUSH;
85 : : else
86 [ + - ]: 1 : ereport(ERROR,
87 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
88 : : errmsg("unrecognized value for %s option \"%s\": \"%s\"",
89 : : "WAIT", defel->defname, mode_str),
90 : : parser_errposition(pstate, defel->location)));
91 : : }
92 [ + + ]: 491 : else if (strcmp(defel->defname, "timeout") == 0)
93 : : {
94 : : char *timeout_str;
95 : : const char *hintmsg;
96 : :
97 [ + + ]: 260 : if (timeout_specified)
98 : 1 : errorConflictingDefElem(defel, pstate);
99 : 259 : timeout_specified = true;
100 : :
101 : 259 : timeout_str = defGetString(defel);
102 : :
103 [ + + ]: 259 : if (!parse_int(timeout_str, &timeout, GUC_UNIT_MS, &hintmsg))
104 [ + - + + ]: 2 : ereport(ERROR,
105 : : errcode(ERRCODE_INVALID_PARAMETER_VALUE),
106 : : errmsg("invalid timeout value: \"%s\"", timeout_str),
107 : : hintmsg ? errhint("%s", _(hintmsg)) : 0,
108 : : parser_errposition(pstate, defel->location));
109 : :
110 [ + + ]: 257 : if (timeout < 0)
111 [ + - ]: 1 : ereport(ERROR,
112 : : errcode(ERRCODE_INVALID_PARAMETER_VALUE),
113 : : errmsg("timeout cannot be negative"),
114 : : parser_errposition(pstate, defel->location));
115 : : }
116 [ + + ]: 231 : else if (strcmp(defel->defname, "no_throw") == 0)
117 : : {
118 [ + + ]: 229 : if (no_throw_specified)
119 : 1 : errorConflictingDefElem(defel, pstate);
120 : :
121 : 228 : no_throw_specified = true;
122 : :
123 : 228 : throw = !defGetBoolean(defel);
124 : : }
125 : : else
126 : : {
127 [ + - ]: 2 : ereport(ERROR,
128 : : errcode(ERRCODE_SYNTAX_ERROR),
129 : : errmsg("option \"%s\" not recognized",
130 : : defel->defname),
131 : : parser_errposition(pstate, defel->location));
132 : : }
133 : : }
134 : :
135 : : /*
136 : : * We are going to wait for the LSN. We should first care that we don't
137 : : * hold a snapshot and correspondingly our MyProc->xmin is invalid.
138 : : * Otherwise, our snapshot could prevent the replay of WAL records
139 : : * implying a kind of self-deadlock. This is the reason why WAIT is a
140 : : * command, not a procedure or function.
141 : : *
142 : : * Non-top-level contexts are rejected above, but be defensive and pop any
143 : : * active snapshot if one is present. PortalRunUtility() can tolerate
144 : : * utility commands that remove the active snapshot.
145 : : */
146 [ - + ]: 277 : if (ActiveSnapshotSet())
147 : 0 : PopActiveSnapshot();
148 : :
149 : : /*
150 : : * At second, invalidate a catalog snapshot if any. And we should be done
151 : : * with the preparation.
152 : : */
153 : 277 : InvalidateCatalogSnapshot();
154 : :
155 : : /* Give up if there is still an active or registered snapshot. */
156 [ + + ]: 277 : if (HaveRegisteredOrActiveSnapshot())
157 [ + - + + ]: 2 : ereport(ERROR,
158 : : errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
159 : : errmsg("WAIT cannot be executed while the current transaction holds a snapshot"),
160 : : IsolationUsesXactSnapshot() ?
161 : : errdetail("This transaction runs at an isolation level higher than READ COMMITTED, so it holds a snapshot from its first query until it ends.") : 0);
162 : :
163 : : /*
164 : : * As the result we should hold no snapshot, and correspondingly our xmin
165 : : * should be unset.
166 : : */
167 : : Assert(MyProc->xmin == InvalidTransactionId);
168 : :
169 : : /*
170 : : * Validate that the requested mode matches the current server state.
171 : : * Primary modes can only be used on a primary.
172 : : */
173 [ + + ]: 275 : if (lsnType == WAIT_LSN_TYPE_PRIMARY_FLUSH)
174 : : {
175 [ + + ]: 14 : if (RecoveryInProgress())
176 [ + - ]: 1 : ereport(ERROR,
177 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
178 : : errmsg("recovery is in progress"),
179 : : errhint("Waiting for primary_flush can only be done on a primary server. "
180 : : "Use standby_flush mode on a standby server.")));
181 : : }
182 : :
183 : : /*
184 : : * Conservatively reject an unsatisfied standby LSN wait while this
185 : : * backend holds a granted heavyweight lock. Recovery may need one of
186 : : * those locks, directly or through another backend, before replay can
187 : : * advance far enough to satisfy our wait. This can create a cycle: we
188 : : * wait for recovery, while recovery waits for us to release the lock.
189 : : *
190 : : * However, we do not register our dependency on WAL progress as a lock
191 : : * wait, so the deadlock detector cannot see the complete cycle. With
192 : : * unlimited recovery-conflict delays and no other timeout or
193 : : * cancellation, the cycle can persist indefinitely.
194 : : *
195 : : * Write and flush waits can also depend on startup. Without an active
196 : : * receiver, their replay floor can be their only source of progress, so
197 : : * holding a lock needed by replay can create the same cycle.
198 : : *
199 : : * Streaming can initially provide independent progress, but reception can
200 : : * stop before the target is reached. Restarting reception requires
201 : : * startup, and stalled replay prevents further advancement of
202 : : * restartpoints used to recycle old WAL, so continued reception can
203 : : * exhaust available space. An active receiver at the start of the wait
204 : : * therefore does not guarantee that the wait can finish while replay
205 : : * remains blocked.
206 : : *
207 : : * Apply the restriction to all standby modes, including some write and
208 : : * flush waits that an active receiver could satisfy while locks remain
209 : : * held. Requests whose target is observed as already reached are exempt
210 : : * from this restriction.
211 : : */
212 [ + + + + ]: 274 : if ((lsnType == WAIT_LSN_TYPE_STANDBY_REPLAY ||
213 [ + + ]: 29 : lsnType == WAIT_LSN_TYPE_STANDBY_WRITE ||
214 [ + + ]: 261 : lsnType == WAIT_LSN_TYPE_STANDBY_FLUSH) &&
215 [ + + ]: 519 : RecoveryInProgress() &&
216 : 258 : lsn > GetCurrentLSNForWaitType(lsnType))
217 : : {
218 : : LOCKTAG locktag;
219 : :
220 [ + + ]: 45 : if (GetAnyGrantedHeavyweightLock(&locktag))
221 : : {
222 : : StringInfoData locktagbuf;
223 : :
224 : 2 : initStringInfo(&locktagbuf);
225 : 2 : DescribeLockTag(&locktagbuf, &locktag);
226 : :
227 [ + - ]: 2 : ereport(ERROR,
228 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
229 : : errmsg("cannot wait for a standby LSN while holding locks"),
230 : : errdetail("This session holds a lock on %s, which could make recovery wait for this session while this session waits for recovery.",
231 : : locktagbuf.data),
232 : : errhint("Release the locks, or execute WAIT before acquiring them.")));
233 : : }
234 : : }
235 : :
236 : : /* Now wait for the LSN */
237 : 272 : waitLSNResult = WaitForLSN(lsnType, lsn, timeout);
238 : :
239 : : /*
240 : : * Process the result of WaitForLSN(). Throw appropriate error if needed.
241 : : */
242 [ + + + - ]: 271 : switch (waitLSNResult)
243 : : {
244 : 260 : case WAIT_LSN_RESULT_SUCCESS:
245 : : /* Nothing to do on success */
246 : 260 : result = "success";
247 : 260 : break;
248 : :
249 : 6 : case WAIT_LSN_RESULT_TIMEOUT:
250 [ + + ]: 6 : if (throw)
251 : : {
252 : 1 : XLogRecPtr currentLSN = GetCurrentLSNForWaitType(lsnType);
253 : :
254 [ + - - - : 1 : switch (lsnType)
- ]
255 : : {
256 : 1 : case WAIT_LSN_TYPE_STANDBY_REPLAY:
257 [ + - ]: 1 : ereport(ERROR,
258 : : errcode(ERRCODE_QUERY_CANCELED),
259 : : errmsg("timed out while waiting for target LSN %X/%08X to be replayed; current standby_replay LSN %X/%08X",
260 : : LSN_FORMAT_ARGS(lsn),
261 : : LSN_FORMAT_ARGS(currentLSN)));
262 : : break;
263 : :
264 : 0 : case WAIT_LSN_TYPE_STANDBY_WRITE:
265 [ # # ]: 0 : ereport(ERROR,
266 : : errcode(ERRCODE_QUERY_CANCELED),
267 : : errmsg("timed out while waiting for target LSN %X/%08X to be written; current standby_write LSN %X/%08X",
268 : : LSN_FORMAT_ARGS(lsn),
269 : : LSN_FORMAT_ARGS(currentLSN)));
270 : : break;
271 : :
272 : 0 : case WAIT_LSN_TYPE_STANDBY_FLUSH:
273 [ # # ]: 0 : ereport(ERROR,
274 : : errcode(ERRCODE_QUERY_CANCELED),
275 : : errmsg("timed out while waiting for target LSN %X/%08X to be flushed; current standby_flush LSN %X/%08X",
276 : : LSN_FORMAT_ARGS(lsn),
277 : : LSN_FORMAT_ARGS(currentLSN)));
278 : : break;
279 : :
280 : 0 : case WAIT_LSN_TYPE_PRIMARY_FLUSH:
281 [ # # ]: 0 : ereport(ERROR,
282 : : errcode(ERRCODE_QUERY_CANCELED),
283 : : errmsg("timed out while waiting for target LSN %X/%08X to be flushed; current primary_flush LSN %X/%08X",
284 : : LSN_FORMAT_ARGS(lsn),
285 : : LSN_FORMAT_ARGS(currentLSN)));
286 : : break;
287 : :
288 : 0 : default:
289 [ # # ]: 0 : elog(ERROR, "unexpected wait LSN type %d", lsnType);
290 : : }
291 : : }
292 : : else
293 : 5 : result = "timeout";
294 : 5 : break;
295 : :
296 : 5 : case WAIT_LSN_RESULT_NOT_IN_RECOVERY:
297 [ + + ]: 5 : if (throw)
298 : : {
299 [ + + ]: 4 : if (PromoteIsTriggered())
300 : : {
301 : 3 : XLogRecPtr currentLSN = GetCurrentLSNForWaitType(lsnType);
302 : :
303 [ + + + - ]: 3 : switch (lsnType)
304 : : {
305 : 1 : case WAIT_LSN_TYPE_STANDBY_REPLAY:
306 [ + - ]: 1 : ereport(ERROR,
307 : : errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
308 : : errmsg("recovery is not in progress"),
309 : : errdetail("Recovery ended before target LSN %X/%08X was replayed; last standby_replay LSN %X/%08X.",
310 : : LSN_FORMAT_ARGS(lsn),
311 : : LSN_FORMAT_ARGS(currentLSN)));
312 : : break;
313 : :
314 : 1 : case WAIT_LSN_TYPE_STANDBY_WRITE:
315 [ + - ]: 1 : ereport(ERROR,
316 : : errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
317 : : errmsg("recovery is not in progress"),
318 : : errdetail("Recovery ended before target LSN %X/%08X was written; last standby_write LSN %X/%08X.",
319 : : LSN_FORMAT_ARGS(lsn),
320 : : LSN_FORMAT_ARGS(currentLSN)));
321 : : break;
322 : :
323 : 1 : case WAIT_LSN_TYPE_STANDBY_FLUSH:
324 [ + - ]: 1 : ereport(ERROR,
325 : : errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
326 : : errmsg("recovery is not in progress"),
327 : : errdetail("Recovery ended before target LSN %X/%08X was flushed; last standby_flush LSN %X/%08X.",
328 : : LSN_FORMAT_ARGS(lsn),
329 : : LSN_FORMAT_ARGS(currentLSN)));
330 : : break;
331 : :
332 : 0 : default:
333 [ # # ]: 0 : elog(ERROR, "unexpected wait LSN type %d", lsnType);
334 : : }
335 : : }
336 : : else
337 : : {
338 [ - - + - ]: 1 : switch (lsnType)
339 : : {
340 : 0 : case WAIT_LSN_TYPE_STANDBY_REPLAY:
341 [ # # ]: 0 : ereport(ERROR,
342 : : errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
343 : : errmsg("recovery is not in progress"),
344 : : errhint("Waiting for the %s LSN can only be executed during recovery.", "standby_replay"));
345 : : break;
346 : :
347 : 0 : case WAIT_LSN_TYPE_STANDBY_WRITE:
348 [ # # ]: 0 : ereport(ERROR,
349 : : errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
350 : : errmsg("recovery is not in progress"),
351 : : errhint("Waiting for the %s LSN can only be executed during recovery.", "standby_write"));
352 : : break;
353 : :
354 : 1 : case WAIT_LSN_TYPE_STANDBY_FLUSH:
355 [ + - ]: 1 : ereport(ERROR,
356 : : errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
357 : : errmsg("recovery is not in progress"),
358 : : errhint("Waiting for the %s LSN can only be executed during recovery.", "standby_flush"));
359 : : break;
360 : :
361 : 0 : default:
362 [ # # ]: 0 : elog(ERROR, "unexpected wait LSN type %d", lsnType);
363 : : }
364 : : }
365 : : }
366 : : else
367 : 1 : result = "not in recovery";
368 : 1 : break;
369 : : }
370 : :
371 : : /* need a tuple descriptor representing a single TEXT column */
372 : 266 : tupdesc = WaitStmtResultDesc(stmt);
373 : :
374 : : /* prepare for projection of tuples */
375 : 266 : tstate = begin_tup_output_tupdesc(dest, tupdesc, &TTSOpsVirtual);
376 : :
377 : : /* Send it */
378 : 266 : do_text_output_oneline(tstate, result);
379 : :
380 : 266 : end_tup_output(tstate);
381 : 266 : }
382 : :
383 : : TupleDesc
384 : 556 : WaitStmtResultDesc(WaitStmt *stmt)
385 : : {
386 : : TupleDesc tupdesc;
387 : :
388 : : /*
389 : : * Need a tuple descriptor representing a single TEXT column.
390 : : *
391 : : * We use TupleDescInitBuiltinEntry instead of TupleDescInitEntry to avoid
392 : : * syscache access. This is important because WaitStmtResultDesc may be
393 : : * called after snapshots have been released, and we must not re-establish
394 : : * a catalog snapshot which could cause recovery conflicts on a standby.
395 : : */
396 : 556 : tupdesc = CreateTemplateTupleDesc(1);
397 : 556 : TupleDescInitBuiltinEntry(tupdesc, (AttrNumber) 1, "status",
398 : : TEXTOID, -1, 0);
399 : 556 : TupleDescFinalize(tupdesc);
400 : 556 : return tupdesc;
401 : : }
|
