mirror of https://github.com/waku-org/nwaku.git
chore: Better postgres duplicate insert (#2535)
* postgres_driver: add ON CONFLICT DO NOTHING in the insert statement * test_driver_postgres: adapt test to ON CONFLICT DO NOTHING The insert does not fail when duplicate, it returns a positive response when doing 'put' of a duplicated row. The test is adapted so that we just check that the number of messages doesn't grow after trying to add a duplicated row.
This commit is contained in:
parent
3ba4378cbe
commit
693a1778d1
|
@ -164,6 +164,9 @@ suite "Postgres driver":
|
||||||
let msg1 = fakeWakuMessage(ts = now)
|
let msg1 = fakeWakuMessage(ts = now)
|
||||||
let msg2 = fakeWakuMessage(ts = now)
|
let msg2 = fakeWakuMessage(ts = now)
|
||||||
|
|
||||||
|
let initialNumMsgs = (await driver.getMessagesCount()).valueOr:
|
||||||
|
raiseAssert "could not get num mgs correctly: " & $error
|
||||||
|
|
||||||
var putRes = await driver.put(
|
var putRes = await driver.put(
|
||||||
DefaultPubsubTopic,
|
DefaultPubsubTopic,
|
||||||
msg1,
|
msg1,
|
||||||
|
@ -173,6 +176,12 @@ suite "Postgres driver":
|
||||||
)
|
)
|
||||||
assert putRes.isOk(), putRes.error
|
assert putRes.isOk(), putRes.error
|
||||||
|
|
||||||
|
var newNumMsgs = (await driver.getMessagesCount()).valueOr:
|
||||||
|
raiseAssert "could not get num mgs correctly: " & $error
|
||||||
|
|
||||||
|
assert newNumMsgs == (initialNumMsgs + 1.int64),
|
||||||
|
"wrong number of messages: " & $newNumMsgs
|
||||||
|
|
||||||
putRes = await driver.put(
|
putRes = await driver.put(
|
||||||
DefaultPubsubTopic,
|
DefaultPubsubTopic,
|
||||||
msg2,
|
msg2,
|
||||||
|
@ -180,4 +189,11 @@ suite "Postgres driver":
|
||||||
computeMessageHash(DefaultPubsubTopic, msg2),
|
computeMessageHash(DefaultPubsubTopic, msg2),
|
||||||
msg2.timestamp,
|
msg2.timestamp,
|
||||||
)
|
)
|
||||||
assert not putRes.isOk()
|
|
||||||
|
assert putRes.isOk()
|
||||||
|
|
||||||
|
newNumMsgs = (await driver.getMessagesCount()).valueOr:
|
||||||
|
raiseAssert "could not get num mgs correctly: " & $error
|
||||||
|
|
||||||
|
assert newNumMsgs == (initialNumMsgs + 1.int64),
|
||||||
|
"wrong number of messages: " & $newNumMsgs
|
||||||
|
|
|
@ -113,14 +113,7 @@ proc handleMessage*(
|
||||||
|
|
||||||
(await self.driver.put(pubsubTopic, msg, msgDigest, msgHash, msgTimestamp)).isOkOr:
|
(await self.driver.put(pubsubTopic, msg, msgDigest, msgHash, msgTimestamp)).isOkOr:
|
||||||
waku_archive_errors.inc(labelValues = [insertFailure])
|
waku_archive_errors.inc(labelValues = [insertFailure])
|
||||||
# Prevent spamming the logs when multiple nodes are connected to the same database.
|
|
||||||
# In that case, the message cannot be inserted but is an expected "insert error"
|
|
||||||
# and therefore we reduce its visibility by having the log in trace level.
|
|
||||||
if "duplicate key value violates unique constraint" in error:
|
|
||||||
trace "failed to insert message", err = error
|
|
||||||
else:
|
|
||||||
debug "failed to insert message", err = error
|
debug "failed to insert message", err = error
|
||||||
|
|
||||||
let insertDuration = getTime().toUnixFloat() - insertStartTime
|
let insertDuration = getTime().toUnixFloat() - insertStartTime
|
||||||
waku_archive_insert_duration_seconds.observe(insertDuration)
|
waku_archive_insert_duration_seconds.observe(insertDuration)
|
||||||
|
|
||||||
|
|
|
@ -31,7 +31,7 @@ type PostgresDriver* = ref object of ArchiveDriver
|
||||||
const InsertRowStmtName = "InsertRow"
|
const InsertRowStmtName = "InsertRow"
|
||||||
const InsertRowStmtDefinition = # TODO: get the sql queries from a file
|
const InsertRowStmtDefinition = # TODO: get the sql queries from a file
|
||||||
"""INSERT INTO messages (id, messageHash, storedAt, contentTopic, payload, pubsubTopic,
|
"""INSERT INTO messages (id, messageHash, storedAt, contentTopic, payload, pubsubTopic,
|
||||||
version, timestamp) VALUES ($1, $2, $3, $4, $5, $6, $7, $8);"""
|
version, timestamp) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) ON CONFLICT DO NOTHING;"""
|
||||||
|
|
||||||
const SelectNoCursorAscStmtName = "SelectWithoutCursorAsc"
|
const SelectNoCursorAscStmtName = "SelectWithoutCursorAsc"
|
||||||
const SelectNoCursorAscStmtDef =
|
const SelectNoCursorAscStmtDef =
|
||||||
|
@ -679,6 +679,7 @@ proc loopPartitionFactory(
|
||||||
debug "creating a new partition for the future"
|
debug "creating a new partition for the future"
|
||||||
## The current used partition is the last one that was created.
|
## The current used partition is the last one that was created.
|
||||||
## Thus, let's create another partition for the future.
|
## Thus, let's create another partition for the future.
|
||||||
|
|
||||||
(
|
(
|
||||||
await self.addPartition(
|
await self.addPartition(
|
||||||
newestPartition.getLastMoment(), PartitionsRangeInterval
|
newestPartition.getLastMoment(), PartitionsRangeInterval
|
||||||
|
|
Loading…
Reference in New Issue