본문 바로가기
WIKI 기술 지식 베이스

Concurrent Writes

원문 보기 위키 갱신

기본 설정의 Turso는 한 번에 한 연결만 쓸 수 있지만, MVCC(Multi-Version Concurrency Control)를 쓰면 여러 연결이 동시에 쓸 수 있어요. 같은 데이터를 두 트랜잭션이 수정하면 한쪽이 충돌 오류를 받고 롤백 후 재시도하게 돼요.

출처: 문서

본문

기본 설정의 Turso는 한 시점에 하나의 연결만 쓰기를 허용해요. MVCC(Multi-Version Concurrency Control)를 사용하면 여러 연결이 동시에 쓸 수 있어요. 두 트랜잭션이 같은 데이터를 수정하면 한쪽은 충돌 오류를 받아 롤백하고 다시 시도해야 해요.

Enable MVCC

데이터베이스 연결에 저널 모드를 설정해요.

PRAGMA journal_mode = 'mvcc';

BEGIN CONCURRENT

BEGIN 대신 BEGIN CONCURRENT를 사용하면 다른 쓰기 작업이 병렬로 진행되는 것을 허용하는 트랜잭션을 시작해요.

BEGIN CONCURRENT;
-- 여기에 쓰기를 작성해요
COMMIT;

두 트랜잭션이 같은 행을 만지면 한쪽이 충돌 오류를 받아 롤백하고 다시 시도해야 해요. 겹치지 않는 데이터에 쓰는 트랜잭션은 충돌 없이 진행돼요.

Handling Conflicts

애플리케이션은 충돌 오류를 감지하고 트랜잭션을 다시 시도해야 해요.

Example

```rust fn is_retryable(e: &Error) -> bool { matches!(e, Error::Busy(_) | Error::BusySnapshot(_)) || matches!(e, Error::Error(msg) if msg.contains("conflict")) } 
#[tokio::main]
async fn main() -> Result<(), Error> {
    let tmp = NamedTempFile::new().expect("failed to create temp file");
    let db = Builder::new_local(tmp.path().to_str().unwrap())
        .build()
        .await?;

    let conn = db.connect()?;
    conn.pragma_update("journal_mode", "'mvcc'").await?;
    conn.execute("CREATE TABLE hits (val INTEGER)", ()).await?;

    let mut handles = Vec::new();
    for _ in 0..16 {
        let db = db.clone();
        handles.push(tokio::spawn(async move {
            let val = rand::rng().random_range(1..=100);
            let conn = db.connect()?;
            loop {
                conn.execute("BEGIN CONCURRENT", ()).await?;
                let result = conn
                    .execute(&format!("INSERT INTO hits VALUES ({val})"), ())
                    .await
                    .and(conn.execute("COMMIT", ()).await);
                match result {
                    Ok(_) => return Ok::<_, Error>(val),
                    Err(ref e) if is_retryable(e) => {
                        let _ = conn.execute("ROLLBACK", ()).await;
                        tokio::task::yield_now().await;
                    }
                    Err(e) => {
                        let _ = conn.execute("ROLLBACK", ()).await;
                        return Err(e);
                    }
                }
            }
        }));
    }

    for handle in handles {
        handle.await.expect("task panicked")?;
    }

    Ok(())
}
```
```js const DB_PATH = join(mkdtempSync(join(tmpdir(), "turso-mvcc-")), "hits.db"); 
function isRetryable(err) {
  const msg = (err?.message ?? "").toLowerCase();
  return msg.includes("conflict") || msg.includes("busy");
}

async function writeWorker() {
  const val = Math.floor(Math.random() * 100) + 1;
  const db = await connect(DB_PATH);
  try {
    await db.exec("PRAGMA journal_mode = 'mvcc'");
    while (true) {
      await db.exec("BEGIN CONCURRENT");
      try {
        await db.exec(`INSERT INTO hits VALUES (${val})`);
        await db.exec("COMMIT");
        return val;
      } catch (err) {
        try { await db.exec("ROLLBACK"); } catch (_) {}
        if (!isRetryable(err)) throw err;
        await new Promise((r) => setImmediate(r));
      }
    }
  } finally {
    await db.close();
  }
}

const setup = await connect(DB_PATH);
await setup.exec("PRAGMA journal_mode = 'mvcc'");
await setup.exec("CREATE TABLE hits (val INTEGER)");
await setup.close();

await Promise.all(Array.from({ length: 16 }, () => writeWorker()));
```
```python def is_retryable(e: Exception) -> bool: msg = str(e).lower() return "conflict" in msg or "busy" in msg 
def write_worker(db_path: str) -> int:
    val = random.randint(1, 100)
    conn = turso.connect(db_path)
    try:
        conn.execute("PRAGMA journal_mode = 'mvcc'").fetchone()
        while True:
            conn.execute("BEGIN CONCURRENT")
            try:
                conn.execute("INSERT INTO hits VALUES (?)", (val,))
                conn.execute("COMMIT")
                return val
            except Exception as e:
                try:
                    conn.execute("ROLLBACK")
                except Exception:
                    pass
                if not is_retryable(e):
                    raise
    finally:
        conn.close()

async def main() -> None:
    with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as tmp:
        db_path = tmp.name

    setup = turso.connect(db_path)
    setup.execute("PRAGMA journal_mode = 'mvcc'").fetchone()
    setup.execute("CREATE TABLE hits (val INTEGER)")
    setup.commit()
    setup.close()

    tasks = [asyncio.to_thread(write_worker, db_path) for _ in range(16)]
    await asyncio.gather(*tasks)

asyncio.run(main())
```

전체 예제를 실행해 보세요. Rust · JavaScript · Python

더 알아보기 (Learn more)