fix: implement batches on top of mysql #85
Conversation
pjenvey
force-pushed
the
feat/51
branch
2 times, most recently
from
143a6fe
to
f7cca8d
Compare
src/db/mysql/batch.rs
Outdated
| // when we convert one to a bigint in milliseconds, the final digit is | ||
| // always zero. But we want to use the lower digits of the batchid for | ||
| // sharding writes via (batchid % num_tables), and leaving it as zero would | ||
| // skew the sharding distribution. |
There was a problem hiding this comment.
Do you in fact do this sharding across multiple "batches" tables?
There was a problem hiding this comment.
Nope, so I removed this comment and the associated code
| .filter(batches::collection_id.eq(&collection_id)) | ||
| .filter(batches::id.eq(¶ms.id)) | ||
| .get_result::<i32>(&db.conn) | ||
| .optional()?; |
There was a problem hiding this comment.
This feels like it should also check whether the batch has expired.
src/db/mysql/batch.rs
Outdated
| } | ||
|
|
||
| pub fn append(db: &MysqlDb, params: params::AppendToBatch) -> Result<()> { | ||
| // XXX: this is possibly an upsert |
There was a problem hiding this comment.
Not sure I agree about an upsert here, you should always know whether you have a new batch or an existing batch.
| .filter(batches::collection_id.eq(&collection_id)) | ||
| .filter(batches::id.eq(¶ms.id)) | ||
| .set(batches::bsos.eq(batches::bsos.concat(&bsos))) | ||
| .execute(&db.conn)?; |
There was a problem hiding this comment.
If the batch has already expired, should this update it or return an error? Also if the specified batch id does not exist, this should likely return an error rather updating zero rows.
There was a problem hiding this comment.
Sounds good, I've added expiry checks here and other places
src/db/mysql/batch.rs
Outdated
| user_id: params.user_id, | ||
| collection: params.collection, | ||
| bsos, | ||
| }) |
There was a problem hiding this comment.
Ideally we should atomically delete the batch after commiting it, to ensure that it can't be committed twice.
src/db/mysql/batch.rs
Outdated
|
|
||
| let batch = get(&db, gb(uid, coll, id))?.unwrap(); | ||
| assert_ne!(batch.bsos, "".to_owned()); | ||
| // XXX: expiry updated? |
There was a problem hiding this comment.
I don't have strong opinions about this, but my first instance is that expiry should not be updated. Otherwise there's no finite bound on the ultimate lifetime of a batch, a misbehaving client could slow-drip add items to a batch and keep it alive indefinitely.
src/db/mysql/batch.rs
Outdated
|
|
||
| pub fn commit(db: &MysqlDb, params: params::CommitBatch) -> Result<results::CommitBatch> { | ||
| let bsos = batch_string_to_bsos(¶ms.batch.bsos)?; | ||
| db.post_bsos_sync(params::PostBsos { |
There was a problem hiding this comment.
Defering the work here to an existing post_bsos_sync method should give you a lot of edge-cases for free, such as proper handling of partial updates and ttl bumping. I like it! Handling those was a source of some complexity in the python code.
| pub fn timestamp(&self) -> SyncTimestamp { | ||
| self.session.borrow().timestamp | ||
| } | ||
|
|
||
| #[cfg(test)] | ||
| pub(super) fn set_timestamp(&self, timestamp: i64) { | ||
| self.session.borrow_mut().timestamp = SyncTimestamp::from_i64(timestamp).unwrap(); | ||
| pub(super) fn with_delta<T, E, F>(&self, delta: i64, f: F) -> std::result::Result<T, E> |
There was a problem hiding this comment.
This is a neat little abstraction!
like go-syncstorage, we just store the bsos' json in a text column,
concatenating batches together w/ newlines
also:
Closes #51, #18