Line data Source code
1 : /*
2 : This file is part of TALER
3 : Copyright (C) 2026 Taler Systems SA
4 :
5 : TALER is free software; you can redistribute it and/or modify it under the
6 : terms of the GNU General Public License as published by the Free Software
7 : Foundation; either version 3, or (at your option) any later version.
8 :
9 : TALER is distributed in the hope that it will be useful, but WITHOUT ANY
10 : WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
11 : A PARTICULAR PURPOSE. See the GNU General Public License for more details.
12 :
13 : You should have received a copy of the GNU General Public License along with
14 : TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
15 : */
16 : /**
17 : * @file exchangedb/iterate_prewires_below_open_shards.c
18 : * @brief Implementation of the iterate_prewires_below_open_shards function for Postgres
19 : * @author Christian Grothoff
20 : */
21 : #include "taler/taler_pq_lib.h"
22 : #include "exchange-database/iterate_prewires_below_open_shards.h"
23 : #include "helper.h"
24 :
25 : /**
26 : * Closure for #prewire_cb().
27 : */
28 : struct PrewireContext
29 : {
30 : /**
31 : * Function to call on each result.
32 : */
33 : TALER_EXCHANGEDB_WirePreparationIterator cb;
34 :
35 : /**
36 : * Closure for @a cb.
37 : */
38 : void *cb_cls;
39 :
40 : /**
41 : * #GNUNET_OK if everything went fine.
42 : */
43 : enum GNUNET_GenericReturnValue status;
44 : };
45 :
46 :
47 : /**
48 : * Invoke the callback for each result.
49 : *
50 : * @param cls a `struct PrewireContext *`
51 : * @param result SQL result
52 : * @param num_results number of rows in @a result
53 : */
54 : static void
55 172 : prewire_cb (void *cls,
56 : PGresult *result,
57 : unsigned int num_results)
58 : {
59 172 : struct PrewireContext *pc = cls;
60 :
61 179 : for (unsigned int i = 0; i < num_results; i++)
62 : {
63 : uint64_t prewire_uuid;
64 : char *wire_method;
65 7 : void *buf = NULL;
66 : size_t buf_size;
67 7 : struct GNUNET_PQ_ResultSpec rs[] = {
68 7 : GNUNET_PQ_result_spec_uint64 ("prewire_uuid",
69 : &prewire_uuid),
70 7 : GNUNET_PQ_result_spec_string ("wire_method",
71 : &wire_method),
72 7 : GNUNET_PQ_result_spec_variable_size ("buf",
73 : &buf,
74 : &buf_size),
75 : GNUNET_PQ_result_spec_end
76 : };
77 :
78 7 : if (GNUNET_OK !=
79 7 : GNUNET_PQ_extract_result (result,
80 : rs,
81 : i))
82 : {
83 0 : GNUNET_break (0);
84 0 : pc->status = GNUNET_SYSERR;
85 0 : return;
86 : }
87 7 : pc->cb (pc->cb_cls,
88 : prewire_uuid,
89 : wire_method,
90 : buf,
91 : buf_size);
92 7 : GNUNET_PQ_cleanup_result (rs);
93 : }
94 : }
95 :
96 :
97 : enum GNUNET_DB_QueryStatus
98 172 : TALER_EXCHANGEDB_iterate_prewires_below_open_shards (
99 : struct TALER_EXCHANGEDB_PostgresContext *pg,
100 : const char *job_name,
101 : uint64_t limit,
102 : TALER_EXCHANGEDB_WirePreparationIterator cb,
103 : void *cb_cls)
104 : {
105 172 : struct GNUNET_PQ_QueryParam params[] = {
106 172 : GNUNET_PQ_query_param_string (job_name),
107 172 : GNUNET_PQ_query_param_uint64 (&limit),
108 : GNUNET_PQ_query_param_end
109 : };
110 172 : struct PrewireContext pc = {
111 : .cb = cb,
112 : .cb_cls = cb_cls,
113 : .status = GNUNET_OK
114 : };
115 : enum GNUNET_DB_QueryStatus qs;
116 :
117 : /* The bound is the start of the lowest shard still open, or, with none
118 : open, the end of the highest shard: new shards are only ever opened
119 : above that, so every row below the bound belongs to a completed shard.
120 : No lease covers those rows, hence SKIP LOCKED keeps two workers from
121 : executing the same one. */
122 172 : PREPARE (pg,
123 : "iterate_prewires_below_open_shards",
124 : "SELECT"
125 : " prewire_uuid"
126 : ",wire_method"
127 : ",buf"
128 : " FROM prewire"
129 : " WHERE prewire_uuid <"
130 : " COALESCE ("
131 : " (SELECT MIN(start_row)"
132 : " FROM work_shards"
133 : " WHERE job_name=$1"
134 : " AND completed=FALSE),"
135 : " (SELECT end_row"
136 : " FROM work_shards"
137 : " WHERE job_name=$1"
138 : " ORDER BY end_row DESC"
139 : " LIMIT 1),"
140 : " 0)"
141 : " AND finished=FALSE"
142 : " AND failed=FALSE"
143 : " ORDER BY prewire_uuid ASC"
144 : " LIMIT $2"
145 : " FOR UPDATE SKIP LOCKED;");
146 172 : qs = GNUNET_PQ_eval_prepared_multi_select (
147 : pg->conn,
148 : "iterate_prewires_below_open_shards",
149 : params,
150 : &prewire_cb,
151 : &pc);
152 172 : if (GNUNET_OK != pc.status)
153 0 : return GNUNET_DB_STATUS_HARD_ERROR;
154 172 : return qs;
155 : }
|
