Repository: launchbadge/sqlx
File: sqlx-sqlite/src/connection/executor.rs (Rust)

Code before:
        'c: 'e,
        E: Execute<'q, Self::Database>,
        'q: 'e,
        E: 'q,
    {
        let sql = query.sql();
        let arguments = match query.take_arguments().map_err(Error::Encode) {
            Ok(arguments) => arguments,
            Err(error) => return future::ready(Err(error)).boxed(),
        };
        let persistent = query.persistent() && arguments.is_some();

        Box::pin(async move {
            let stream = self
                .worker
                .execute(sql, arguments, self.row_channel_size, persistent)
                .map_ok(flume::Receiver::into_stream)
                .try_flatten_stream();

            futures_util::pin_mut!(stream);

            let mut out = Ok(None);

            while let Some(res) = stream.try_next().await? {
                if let Either::Right(row) = res {
                    out = Ok(Some(row));
                }
            }

            out
        })
    }

    fn prepare_with<'e, 'q: 'e>(
        self,
        sql: &'q str,
        _parameters: &[SqliteTypeInfo],
    ) -> BoxFuture<'e, Result<SqliteStatement<'q>, Error>>
    where
        'c: 'e,
    {
        Box::pin(async move {
            let statement = self.worker.prepare(sql).await?;

            Ok(SqliteStatement {
                sql: sql.into(),
                ..statement
            })
        })
    }


Review comment (bug):
The reason it doesn't work this way is because we didn't want the application to have to wait for the full result set before returning. For example, if someone forgot a `LIMIT 1` on their query and was doing a `SELECT ... ORDER BY ...` to get the first/last item in an ordered set, the call would have to churn through the whole dataset before returning.

The bug happens because of a race condition: the worker sends the row data and then continues calling `sqlite3_step()` and only stops when it notices the result channel is closed, then it calls `sqlite3_reset()` which is supposed to actually commit the changes. In the meantime, the application receives the row and then continues to the next query, which it executes on another connection which may not be able to see the changes yet.

In reality, this needs to be handled proactively by the driver. The message to the worker should specify that it's only expecting one row, and then the worker should step until it gets the row and immediately call `sqlite3_reset()` (by dropping `ExecuteIter`) before returning it. That should fix the race condition.

Code after:
        'c: 'e,
        E: Execute<'q, Self::Database>,
        'q: 'e,
        E: 'q,
    {
        let sql = query.sql();
        let arguments = match query.take_arguments().map_err(Error::Encode) {
            Ok(arguments) => arguments,
            Err(error) => return future::ready(Err(error)).boxed(),
        };
        let persistent = query.persistent() && arguments.is_some();

        Box::pin(async move {
            let stream = self
                .worker
                .execute(sql, arguments, self.row_channel_size, persistent, Some(1))
                .map_ok(flume::Receiver::into_stream)
                .try_flatten_stream();

            futures_util::pin_mut!(stream);

            while let Some(res) = stream.try_next().await? {
                if let Either::Right(row) = res {
                    return Ok(Some(row));
                }
            }

            Ok(None)
        })
    }

    fn prepare_with<'e, 'q: 'e>(
        self,
        sql: &'q str,
        _parameters: &[SqliteTypeInfo],
    ) -> BoxFuture<'e, Result<SqliteStatement<'q>, Error>>
    where
        'c: 'e,
    {
        Box::pin(async move {
            let statement = self.worker.prepare(sql).await?;

            Ok(SqliteStatement {
                sql: sql.into(),
                ..statement
            })
        })
    }

    #[doc(hidden)]
    fn describe<'e, 'q: 'e>(self, sql: &'q str) -> BoxFuture<'e, Result<Describe<Sqlite>, Error>>