Skip to content
rachel/workerd
File

Blob: src/workerd/api/sql.c++

16.1 KB
1// Copyright (c) 2017-2022 Cloudflare, Inc.
2// Licensed under the Apache 2.0 license found in the LICENSE file or at:
3//     https://opensource.org/licenses/Apache-2.0
4 
5#include "sql.h"
6 
7#include "actor-state.h"
8 
9#include <workerd/io/io-context.h>
10#include <workerd/util/autogate.h>
11#include <workerd/util/sentry.h>
12 
13#if _WIN32
14#define strncasecmp _strnicmp
15#else
16#include <strings.h>
17#endif
18 
19namespace workerd::api {
20 
21// Maximum total size of all cached statements (measured in size of the SQL code). If cached
22// statements exceed this, we remove the LRU statement(s).
23//
24// Hopefully most apps don't ever hit this, but it's important to have a limit in case of
25// queries containing dynamic content or excessively large one-off queries.
26static constexpr uint SQL_STATEMENT_CACHE_MAX_SIZE = 1024 * 1024;
27 
28SqlStorage::SqlStorage(jsg::Ref<DurableObjectStorage> storage)
29    : storage(kj::mv(storage)),
30      statementCache(IoContext::current().addObject(kj::heap<StatementCache>())) {}
31 
32SqlStorage::~SqlStorage() {}
33 
34jsg::Ref<SqlStorage::Cursor> SqlStorage::exec(
35    jsg::Lock& js, jsg::JsString querySql, jsg::Arguments<BindingValue> bindings) {
36  auto& context = IoContext::current();
37  TraceContext traceContext = context.makeUserTraceSpan("durable_object_storage_exec"_kjc);
38  traceContext.setTag("db.system.name"_kjc, "cloudflare-durable-object-sql"_kjc);
39  traceContext.setTag("db.operation.name"_kjc, "exec"_kjc);
40  traceContext.setTag("db.query.text"_kjc, kj::str(querySql));
41  traceContext.setTag(
42      "cloudflare.durable_object.query.bindings"_kjc, static_cast<int64_t>(bindings.size()));
43 
44  // Internalize the string, so that the cache can be keyed by string identity rather than content.
45  // Any string we put into the cache is expected to live there for a while anyway, so even if it
46  // is a one-off, internalizing it (which moves it to the old generation) shouldn't hurt.
47  querySql = querySql.internalize(js);
48 
49  auto& db = getDb(js);
50  auto& statementCache = *this->statementCache;
51 
52  kj::Rc<CachedStatement>& slot = statementCache.map.findOrCreate(querySql, [&]() {
53    auto result = kj::rc<CachedStatement>(js, *this, db, querySql, js.toString(querySql));
54    statementCache.totalSize += result->statementSize;
55    return result;
56  });
57 
58  // Move cached statement to end of LRU queue.
59  if (slot->lruLink.isLinked()) {
60    statementCache.lru.remove(*slot.get());
61  }
62  statementCache.lru.add(*slot.get());
63 
64  // In order to get accurate statistics, we have to keep the spans around until the query is
65  // actually done, which for read queries that iterate over a cursor won't be until later.
66  kj::Maybe<kj::Function<void(Cursor&)>> doneCallback;
67  if (traceContext.isObserved()) {
68    doneCallback = [traceContext = context.addObject(kj::heap(kj::mv(traceContext)))](
69                       Cursor& cursor) mutable {
70      int64_t rowsRead = cursor.getRowsRead();
71      int64_t rowsWritten = cursor.getRowsWritten();
72      traceContext->setTag("cloudflare.durable_object.response.rows_read"_kjc, rowsRead);
73      traceContext->setTag("cloudflare.durable_object.response.rows_written"_kjc, rowsWritten);
74    };
75  }
76 
77  if (slot->isShared()) {
78    // Oops, this CachedStatement is currently in-use (presumably by a Cursor).
79    //
80    // SQLite only allows one instance of a statement to run at a time, so we will have to compile
81    // the statement again as a one-off.
82    //
83    // In theory we could try to cache multiple copies of the statement, but as this is probably
84    // exceedingly rare, it is not worth the added code complexity.
85    SqliteDatabase::Regulator& regulator = *this;
86    return js.alloc<Cursor>(
87        js, kj::mv(doneCallback), db, regulator, js.toString(querySql), kj::mv(bindings));
88  }
89 
90  auto result = js.alloc<Cursor>(js, kj::mv(doneCallback), slot.addRef(), kj::mv(bindings));
91 
92  // If the statement cache grew too big, drop the least-recently-used entry.
93  while (statementCache.totalSize > SQL_STATEMENT_CACHE_MAX_SIZE) {
94    auto& toRemove = *statementCache.lru.begin();
95    auto oldQuery = jsg::JsString(toRemove.query.getHandle(js));
96    statementCache.totalSize -= toRemove.statementSize;
97    statementCache.lru.remove(toRemove);
98    KJ_ASSERT(statementCache.map.eraseMatch(oldQuery));
99  }
100 
101  return result;
102}
103 
104SqlStorage::IngestResult SqlStorage::ingest(jsg::Lock& js, kj::String querySql) {
105  auto& context = IoContext::current();
106  TraceContext traceContext = context.makeUserTraceSpan("durable_object_storage_ingest"_kjc);
107  SqliteDatabase::Regulator& regulator = *this;
108  auto result = getDb(js).ingestSql(regulator, querySql);
109 
110  traceContext.setTag(
111      "cloudflare.durable_object.response.rows_read"_kjc, static_cast<int64_t>(result.rowsRead));
112  traceContext.setTag("cloudflare.durable_object.response.rows_written"_kjc,
113      static_cast<int64_t>(result.rowsWritten));
114  traceContext.setTag("cloudflare.durable_object.response.statement_count"_kjc,
115      static_cast<int64_t>(result.statementCount));
116 
117  return IngestResult(
118      kj::str(result.remainder), result.rowsRead, result.rowsWritten, result.statementCount);
119}
120 
121void SqlStorage::setMaxPageCountForTest(jsg::Lock& js, int count) {
122  auto& db = getDb(js);
123  db.run({.regulator = SqliteDatabase::TRUSTED}, kj::str("PRAGMA max_page_count = ", count));
124}
125 
126jsg::Ref<SqlStorage::Statement> SqlStorage::prepare(jsg::Lock& js, jsg::JsString query) {
127  return js.alloc<Statement>(js, JSG_THIS, query);
128}
129 
130double SqlStorage::getDatabaseSize(jsg::Lock& js) {
131  auto& context = IoContext::current();
132  TraceContext traceContext =
133      context.makeUserTraceSpan("durable_object_storage_getDatabaseSize"_kjc);
134  traceContext.setTag("db.operation.name"_kjc, "getDatabaseSize"_kjc);
135  auto& db = getDb(js);
136  int64_t pages = execMemoized(db, pragmaPageCount,
137      "select (select * from pragma_page_count) - (select * from pragma_freelist_count);")
138                      .getInt64(0);
139  auto dbSize = pages * getPageSize(db);
140  traceContext.setTag(
141      "cloudflare.durable_object.response.db_size"_kjc, static_cast<int64_t>(dbSize));
142  return dbSize;
143}
144 
145bool SqlStorage::isAllowedName(kj::StringPtr name) const {
146  if (util::Autogate::isEnabled(util::AutogateKey::SQL_RESTRICT_RESERVED_NAMES)) {
147    return strncasecmp(name.begin(), "_cf_", 4) != 0;
148  }
149  if (name.size() >= 4 && strncasecmp(name.begin(), "_cf_", 4) == 0) {
150    LOG_WARNING_PERIODICALLY("SQL identifier matches reserved _cf_ prefix case-insensitively");
151  }
152  return !name.startsWith("_cf_");
153}
154 
155bool SqlStorage::isAllowedTrigger(kj::StringPtr name) const {
156  return true;
157}
158 
159void SqlStorage::onError(kj::Maybe<int> sqliteErrorCode, kj::StringPtr message) const {
160  JSG_ASSERT(false, Error, message);
161}
162 
163bool SqlStorage::allowTransactions() const {
164  JSG_FAIL_REQUIRE(Error,
165      "To execute a transaction, please use the state.storage.transaction() or "
166      "state.storage.transactionSync() APIs instead of the SQL BEGIN TRANSACTION or SAVEPOINT "
167      "statements. The JavaScript API is safer because it will automatically roll back on "
168      "exceptions, and because it interacts correctly with Durable Objects' automatic atomic "
169      "write coalescing.");
170}
171 
172bool SqlStorage::shouldAddQueryStats() const {
173  // Bill for queries executed from JavaScript.
174  return true;
175}
176 
177SqlStorage::StatementCache::~StatementCache() noexcept(false) {
178  for (auto& entry: lru) {
179    lru.remove(entry);
180  }
181}
182 
183jsg::JsValue SqlStorage::wrapSqlValue(jsg::Lock& js, SqlValue value) {
184  KJ_IF_SOME(v, value) {
185    KJ_SWITCH_ONEOF(v) {
186      KJ_CASE_ONEOF(bytes, kj::Array<byte>) {
187        return jsg::JsValue(js.wrapBytes(kj::mv(bytes)));
188      }
189      KJ_CASE_ONEOF(text, kj::StringPtr) {
190        return js.str(text);
191      }
192      KJ_CASE_ONEOF(number, double) {
193        return js.num(number);
194      }
195    }
196    KJ_UNREACHABLE;
197  } else {
198    return js.null();
199  }
200}
201 
202SqlStorage::Cursor::State::State(SqliteDatabase& db,
203    SqliteDatabase::Regulator& regulator,
204    kj::StringPtr sqlCode,
205    kj::Array<BindingValue> bindingsParam)
206    : bindings(kj::mv(bindingsParam)),
207      query(db.run({.regulator = regulator}, sqlCode, mapBindings(bindings).asPtr())) {}
208 
209SqlStorage::Cursor::State::State(
210    kj::Rc<CachedStatement> cachedStatementParam, kj::Array<BindingValue> bindingsParam)
211    : bindings(kj::mv(bindingsParam)),
212      query(cachedStatement.emplace(kj::mv(cachedStatementParam))
213                ->statement.run(mapBindings(bindings).asPtr())) {}
214 
215SqlStorage::Cursor::~Cursor() noexcept(false) {
216  // If this Cursor was created from a Statement, clear the Statement's currentCursor weak ref.
217  KJ_IF_SOME(s, selfRef) {
218    KJ_IF_SOME(p, s) {
219      if (&p == this) {
220        s = kj::none;
221      }
222    }
223  }
224}
225 
226void SqlStorage::Cursor::initColumnNames(jsg::Lock& js, State& stateRef) {
227  KJ_IF_SOME(cached, stateRef.cachedStatement) {
228    reusedCachedQuery = cached->useCount++ > 0;
229  }
230 
231  js.withinHandleScope([&]() {
232    v8::LocalVector<v8::Value> vec(js.v8Isolate);
233    for (auto i: kj::zeroTo(stateRef.query.columnCount())) {
234      vec.push_back(js.str(stateRef.query.getColumnName(i)));
235    }
236    auto array = jsg::JsArray(v8::Array::New(js.v8Isolate, vec.data(), vec.size()));
237    columnNames = jsg::JsRef<jsg::JsArray>(js, array);
238  });
239}
240 
241double SqlStorage::Cursor::getRowsRead() {
242  KJ_IF_SOME(st, state) {
243    return static_cast<double>(st->query.getRowsRead());
244  } else {
245    return static_cast<double>(rowsRead);
246  }
247}
248 
249double SqlStorage::Cursor::getRowsWritten() {
250  KJ_IF_SOME(st, state) {
251    return static_cast<double>(st->query.getRowsWritten());
252  } else {
253    return static_cast<double>(rowsWritten);
254  }
255}
256 
257SqlStorage::Cursor::RowIterator::Next SqlStorage::Cursor::next(jsg::Lock& js) {
258  auto self = JSG_THIS;
259  auto maybeRow = rowIteratorNext(js, self);
260  bool done = maybeRow == kj::none;
261  return {
262    .done = done,
263    .value = kj::mv(maybeRow),
264  };
265}
266 
267jsg::JsArray SqlStorage::Cursor::toArray(jsg::Lock& js) {
268  auto self = JSG_THIS;
269  v8::LocalVector<v8::Value> results(js.v8Isolate);
270  for (;;) {
271    auto maybeRow = rowIteratorNext(js, self);
272    KJ_IF_SOME(row, maybeRow) {
273      results.push_back(row);
274    } else {
275      break;
276    }
277  }
278 
279  return jsg::JsArray(v8::Array::New(js.v8Isolate, results.data(), results.size()));
280}
281 
282jsg::JsValue SqlStorage::Cursor::one(jsg::Lock& js) {
283  auto self = JSG_THIS;
284  auto result = JSG_REQUIRE_NONNULL(rowIteratorNext(js, self), Error,
285      "Expected exactly one result from SQL query, but got no results.");
286 
287  KJ_IF_SOME(s, state) {
288    // It appears that the query had more results, otherwise we would have set `state` to `none`
289    // inside `iteratorImpl()`.
290    endQuery(*s);
291    JSG_FAIL_REQUIRE(
292        Error, "Expected exactly one result from SQL query, but got multiple results.");
293  }
294 
295  return result;
296}
297 
298jsg::Ref<SqlStorage::Cursor::RowIterator> SqlStorage::Cursor::rows(jsg::Lock& js) {
299  return js.alloc<RowIterator>(JSG_THIS);
300}
301 
302kj::Maybe<jsg::JsObject> SqlStorage::Cursor::rowIteratorNext(jsg::Lock& js, jsg::Ref<Cursor>& obj) {
303  KJ_IF_SOME(values, iteratorImpl(js, obj)) {
304    auto names = obj->columnNames.getHandle(js);
305    jsg::JsObject result = js.obj();
306    KJ_ASSERT(names.size() == values.size());
307    for (auto i: kj::zeroTo(names.size())) {
308      result.set(js, names.get(js, i), jsg::JsValue(values[i]));
309    }
310    return result;
311  } else {
312    return kj::none;
313  }
314}
315 
316jsg::Ref<SqlStorage::Cursor::RawIterator> SqlStorage::Cursor::raw(jsg::Lock& js) {
317  return js.alloc<RawIterator>(JSG_THIS);
318}
319 
320// Returns the set of column names for the current Cursor. An exception will be thrown if the
321// iterator has already been fully consumed. The resulting columns may contain duplicate entries,
322// for instance a `SELECT *` across a join of two tables that share a column name.
323jsg::JsArray SqlStorage::Cursor::getColumnNames(jsg::Lock& js) {
324  return columnNames.getHandle(js);
325}
326 
327kj::Maybe<jsg::JsArray> SqlStorage::Cursor::rawIteratorNext(jsg::Lock& js, jsg::Ref<Cursor>& obj) {
328  KJ_IF_SOME(values, iteratorImpl(js, obj)) {
329    return jsg::JsArray(v8::Array::New(js.v8Isolate, values.data(), values.size()));
330  } else {
331    return kj::none;
332  }
333}
334 
335kj::Maybe<v8::LocalVector<v8::Value>> SqlStorage::Cursor::iteratorImpl(
336    jsg::Lock& js, jsg::Ref<Cursor>& obj) {
337  auto& state = *KJ_UNWRAP_OR(obj->state, {
338    if (obj->canceled) {
339      JSG_FAIL_REQUIRE(Error,
340          "SQL cursor was closed because the same statement was executed again. If you need to "
341          "run multiple copies of the same statement concurrently, you must create multiple "
342          "prepared statement objects.");
343    } else {
344      // Query already done.
345      return kj::none;
346    }
347  });
348 
349  auto& query = state.query;
350 
351  if (query.isDone()) {
352    obj->endQuery(state);
353    return kj::none;
354  }
355 
356  auto n = query.columnCount();
357  v8::LocalVector<v8::Value> results(js.v8Isolate);
358  results.reserve(n);
359  for (auto i: kj::zeroTo(n)) {
360    SqlValue value;
361    KJ_SWITCH_ONEOF(query.getValue(i)) {
362      KJ_CASE_ONEOF(data, kj::ArrayPtr<const byte>) {
363        value.emplace(kj::heapArray(data));
364      }
365      KJ_CASE_ONEOF(text, kj::StringPtr) {
366        value.emplace(text);
367      }
368      KJ_CASE_ONEOF(i, int64_t) {
369        // int64 will become BigInt, but most applications won't want all their integers to be
370        // BigInt. We will coerce to a double here.
371        // TODO(someday): Allow applications to request that certain columns use BigInt.
372        value.emplace(static_cast<double>(i));
373      }
374      KJ_CASE_ONEOF(d, double) {
375        value.emplace(d);
376      }
377      KJ_CASE_ONEOF(_, decltype(nullptr)) {
378        // leave value null
379      }
380    }
381    results.push_back(wrapSqlValue(js, kj::mv(value)));
382  }
383 
384  // Proactively iterate to the next row and, if it turns out the query is done, discard it. This
385  // is an optimization to make sure that the statement can be returned to the statement cache once
386  // the application has iterated over all results, even if the application fails to call next()
387  // one last time to get `{done: true}`. A common case where this could happen is if the app is
388  // expecting zero or one results, so it calls `exec(...).next()`. In the case that one result
389  // was returned, the application may not bother calling `next()` again. If we hadn't proactively
390  // iterated ahead by one, then the statement would not be returned to the cache until it was
391  // GC'ed, which might prevent the cache from being effective in the meantime.
392  //
393  // Unfortunately, this does not help with the case where the application stops iterating with
394  // results still available from the cursor. There's not much we can do about that case since
395  // there's no way to know if the app might come back and try to use the cursor again later.
396  query.nextRow();
397  if (query.isDone()) {
398    obj->endQuery(state);
399  }
400 
401  return kj::mv(results);
402}
403 
404void SqlStorage::Cursor::endQuery(State& stateRef) {
405  // Save off row counts before the query goes away.
406  rowsRead = stateRef.query.getRowsRead();
407  rowsWritten = stateRef.query.getRowsWritten();
408 
409  KJ_IF_SOME(cb, doneCallback) {
410    cb(*this);
411    doneCallback = kj::none;
412  }
413 
414  // Clean up the query proactively.
415  state = kj::none;
416}
417 
418kj::Array<const SqliteDatabase::Query::ValuePtr> SqlStorage::Cursor::mapBindings(
419    kj::ArrayPtr<BindingValue> values) {
420  return KJ_MAP(value, values) -> SqliteDatabase::Query::ValuePtr {
421    KJ_IF_SOME(v, value) {
422      KJ_SWITCH_ONEOF(v) {
423        KJ_CASE_ONEOF(data, kj::Array<const byte>) {
424          return data.asPtr();
425        }
426        KJ_CASE_ONEOF(text, kj::String) {
427          return text.asPtr();
428        }
429        KJ_CASE_ONEOF(d, double) {
430          return d;
431        }
432      }
433    } else {
434      return nullptr;
435    }
436    KJ_UNREACHABLE;
437  };
438}
439 
440jsg::Ref<SqlStorage::Cursor> SqlStorage::Statement::run(
441    jsg::Lock& js, jsg::Arguments<BindingValue> bindings) {
442  return sqlStorage->exec(js, jsg::JsString(query.getHandle(js)), kj::mv(bindings));
443}
444 
445void SqlStorage::visitForMemoryInfo(jsg::MemoryTracker& tracker) const {
446  tracker.trackField("storage", storage);
447  tracker.trackFieldWithSize("IoPtr<SqliteDatabase>", sizeof(IoPtr<SqliteDatabase>));
448  if (pragmaPageCount != kj::none) {
449    tracker.trackFieldWithSize(
450        "IoPtr<SqllitDatabase::Statement>", sizeof(IoPtr<SqliteDatabase::Statement>));
451  }
452  if (pragmaGetMaxPageCount != kj::none) {
453    tracker.trackFieldWithSize(
454        "IoPtr<SqllitDatabase::Statement>", sizeof(IoPtr<SqliteDatabase::Statement>));
455  }
456}
457 
458}  // namespace workerd::api