feat(cogs): report backend changes to a change stream - #595
Conversation
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## matth/storage-inventory-tracker-2 #595 +/- ##
=====================================================================
+ Coverage 88.24% 88.46% +0.21%
=====================================================================
Files 104 104
Lines 16803 17055 +252
=====================================================================
+ Hits 14828 15087 +259
+ Misses 1975 1968 -7
☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
ab70a01 to
959eb00
Compare
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes and found 1 potential issue.
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit 959eb00. Configure here.
959eb00 to
6d8e991
Compare
![sentry[bot]](https://avatars.githubusercontent.com/in/12637?s=60&v=4){kind=small}
| Ok(()) | ||
| }) | ||
| .await | ||
| .await?; |
There was a problem hiding this comment.
Bug: delete_object for the GCS backend reports a delete event to the change stream even when the object does not exist (receives a 404), creating spurious records.
Severity: MEDIUM
Suggested Fix
Modify the delete_object function in the GCS backend to only call self.change_stream.delete(id) if the delete operation was successful (i.e., did not result in a 404 status). This could be achieved by having the with_retry closure return a boolean indicating whether a delete occurred, and then conditionally calling the change stream method.
Prompt for AI Agent
Review the code at the location below. A potential bug has been identified by an AI
agent. Verify if this is a real issue. If it is, propose a fix; if not, explain why it's
not valid.
Location: objectstore-service/src/backend/gcs.rs#L900
Potential issue: In the GCS backend's `delete_object` implementation, if a delete
operation is attempted on a non-existent object, the GCS API returns a `404 NOT FOUND`
status. The code correctly handles this status to avoid an error but then proceeds to
unconditionally call `self.change_stream.delete(id)`. This results in a delete event
being logged to the change stream for an object that was never deleted, leading to
inaccurate inventory tracking and potentially incorrect cost attribution.
Did we get this right? 👍 / 👎 to inform future reviews.
There was a problem hiding this comment.
this is harmless outside of the incredibly unlikely case of a racing delete/write being sequenced differently in Objectstore and Kafka. but i will fix
There was a problem hiding this comment.
Easy to fix by returning a boolean from the retry closure.
6d8e991 to
28100c2
Compare
| /// | ||
| /// This function does not distinguish between object rows and tombstone rows. It does not | ||
| /// include Bigtable's own overhead. | ||
| fn row_bytes(path: &[u8], mutations: &[v2::Mutation]) -> u64 { |
There was a problem hiding this comment.
nit: Let's use length or size since bytes is commonly used to refer to the actual data.
| // Inline `put_row()` because we need the mutations to compute their size. | ||
| let mutations = object_mutations(metadata.clone(), payload.into_bytes().into())?; | ||
| self.mutate(path.clone(), mutations.clone(), "put").await?; | ||
| self.report_write(id, &path, &mutations, metadata.time_expires); |
There was a problem hiding this comment.
You can make object_mutations instead return the size alongside the mutations, return that from put_row, and then use the returned value to report the write. This yields all data in the correct sequence, removes the inlining, and mutations no longer have to be cloned.
| Ok(()) | ||
| }) | ||
| .await | ||
| .await?; |
There was a problem hiding this comment.
Easy to fix by returning a boolean from the retry closure.
| // The payload arrives as a stream with no declared length, so the stored size is | ||
| // only known once the upload has drained. Safe from double-counting because this | ||
| // request is not retried. | ||
| let (stored_size, stream) = counting_stream(stream); |
There was a problem hiding this comment.
Please double-check, but GCS should be returning a X-Goog-Stored-Content-Length from the upload request which we can use. Even better, this is authoritative.
Just know that this doesn't include metadata, and GCS charges for metadata as bytes, too.
There was a problem hiding this comment.
TODO also count metadata
2 participants
rig up GCS and Bigtable backends to emit change stream records. nothing is actually plugging in a change stream yet, that'll be future PRs
didn't do other backends yet because i wanted feedback on the stack so far before writing that much more code haha