logos-messaging/storenode-messages-counter
1
0
Fork 0
mirror of https://github.com/logos-messaging/storenode-messages-counter.git synced 2026-01-02 14:13:11 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: update on insert

Browse Source
This commit is contained in:
Richard Ramos 2024-09-17 10:38:14 -04:00
parent c80a625423
commit 8510aef48d
No known key found for this signature in database
GPG Key ID: 1CE87DB518195760
4 changed files with 12 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
cmd/storemsgcounter/execute.go
View File

@ -475,7 +475,11 @@ func (app *Application) fetchStoreNodeMessages(ctx context.Context, runId string
success := false
count := 1
for retry && count <= maxAttempts {
queryLogger.Info("retrieving message history for topic", zap.Int("attempt", count))
requestID := protocol.GenerateRequestID()
queryLogger.Info("retrieving message history for topic",
zap.Int("attempt", count),
zap.String("requestID", hex.EncodeToString(requestID)))
tCtx, cancel := context.WithTimeout(ctx, 1*time.Minute)
result, err = app.node.Store().Query(tCtx, store.FilterCriteria{
@ -486,7 +490,9 @@ func (app *Application) fetchStoreNodeMessages(ctx context.Context, runId string
cancel()
if err != nil {
queryLogger.Error("could not query storenode", zap.Error(err), zap.Int("attempt", count))
queryLogger.Error("could not query storenode",
zap.Error(err), zap.Int("attempt", count),
zap.String("requestID", hex.EncodeToString(requestID)))
time.Sleep(2 * time.Second)
} else {
queryLogger.Info("messages available", zap.Int("len", len(result.Messages())))

2
go.mod
View File

@ -18,7 +18,7 @@ require (
github.com/multiformats/go-multiaddr v0.12.4
github.com/prometheus/client_golang v1.19.1
github.com/urfave/cli/v2 v2.27.2
github.com/waku-org/go-waku v0.8.1-0.20240806200653-4f1d692413e9
github.com/waku-org/go-waku v0.8.1-0.20240917141301-991e872de95c
go.opencensus.io v0.24.0
go.uber.org/zap v1.27.0
golang.org/x/exp v0.0.0-20240506185415-9bf2ced13842

2
go.sum
View File

@ -826,6 +826,8 @@ github.com/waku-org/go-libp2p-rendezvous v0.0.0-20240110193335-a67d1cc760a0 h1:R
github.com/waku-org/go-libp2p-rendezvous v0.0.0-20240110193335-a67d1cc760a0/go.mod h1:EhZP9fee0DYjKH/IOQvoNSy1tSHp2iZadsHGphcAJgY=
github.com/waku-org/go-waku v0.8.1-0.20240806200653-4f1d692413e9 h1:aTOUQm0kKtHiqraFpqj1Ja++C+qyZyeiSPKtXe3Ctac=
github.com/waku-org/go-waku v0.8.1-0.20240806200653-4f1d692413e9/go.mod h1:VNbVmh5UYg3vIvhGV4hCw8QEykq3RScDACo2Y2dIFfg=
github.com/waku-org/go-waku v0.8.1-0.20240917141301-991e872de95c h1:dyJ0JOZjQNNXtOCDNfN7tjpCZKTJLOVOrL5zNbJ/4zQ=
github.com/waku-org/go-waku v0.8.1-0.20240917141301-991e872de95c/go.mod h1:VNbVmh5UYg3vIvhGV4hCw8QEykq3RScDACo2Y2dIFfg=
github.com/waku-org/go-zerokit-rln v0.1.14-0.20240102145250-fa738c0bdf59 h1:jisj+OCI6QydLtFq3Pyhu49wl9ytPN7oAHjMfepHDrA=
github.com/waku-org/go-zerokit-rln v0.1.14-0.20240102145250-fa738c0bdf59/go.mod h1:1PdBdPzyTaKt3VnpAHk3zj+r9dXPFOr3IHZP9nFle6E=
github.com/waku-org/go-zerokit-rln-apple v0.0.0-20230916172309-ee0ee61dde2b h1:KgZVhsLkxsj5gb/FfndSCQu6VYwALrCOgYI3poR95yE=

2
internal/persistence/database.go
View File

@ -284,7 +284,7 @@ func (d *DBStore) UpdateTopicSyncState(tx *sql.Tx, topic string, lastSyncTimesta
}
func (d *DBStore) RecordMessage(uuid string, tx *sql.Tx, msgHash pb.MessageHash, topic string, storenodes []peer.ID, status string) error {
stmt, err := tx.Prepare("INSERT INTO missingMessages(runId, fleet, clusterId, pubsubTopic, messageHash, storenode, msgStatus, storedAt) VALUES ($1, $2, $3, $4, $5, $6, $7, $8)")
stmt, err := tx.Prepare("INSERT INTO missingMessages(runId, fleet, clusterId, pubsubTopic, messageHash, storenode, msgStatus, storedAt) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) ON CONFLICT (messageHash, storenode, fleet) DO UPDATE pubsubTopic = EXCLUDED.pubsubTopic, msgStatus = EXCLUDED.msgStatus, storedAt = EXCLUDED.storedAt, clusterId = EXCLUDED.clusterId")
if err != nil {
return err
}