sigb.us/dendrite
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases 1 Wiki Activity

Actually store EDUs once we retrieved from the database (#2651)

Browse source 
We now actually cache the EDUs once we got them from the database and ensures we only evict them if we successfully deleted them.
This commit is contained in:
Till 2022-08-18 12:14:42 +02:00 committed by GitHub
parent 59bc0a6f4e
commit 7484689ad1
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 15 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
federationapi/storage/shared/storage_edus.go
View file

@ -110,6 +110,7 @@ func (d *Database) GetPendingEDUs(
return fmt.Errorf("json.Unmarshal: %w", err) return fmt.Errorf("json.Unmarshal: %w", err)
} }
edus[&Receipt{nid}] = &event edus[&Receipt{nid}] = &event
d.Cache.StoreFederationQueuedEDU(nid, &event)
} }
return nil return nil
@ -177,20 +178,18 @@ func (d *Database) GetPendingEDUServerNames(
return d.FederationQueueEDUs.SelectQueueEDUServerNames(ctx, nil) return d.FederationQueueEDUs.SelectQueueEDUServerNames(ctx, nil)
} }
// DeleteExpiredEDUs deletes expired EDUs // DeleteExpiredEDUs deletes expired EDUs and evicts them from the cache.
func (d *Database) DeleteExpiredEDUs(ctx context.Context) error { func (d *Database) DeleteExpiredEDUs(ctx context.Context) error {
return d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error { var jsonNIDs []int64
err := d.Writer.Do(d.DB, nil, func(txn *sql.Tx) (err error) {
expiredBefore := gomatrixserverlib.AsTimestamp(time.Now()) expiredBefore := gomatrixserverlib.AsTimestamp(time.Now())
jsonNIDs, err := d.FederationQueueEDUs.SelectExpiredEDUs(ctx, txn, expiredBefore) jsonNIDs, err = d.FederationQueueEDUs.SelectExpiredEDUs(ctx, txn, expiredBefore)
if err != nil { if err != nil {
return err return err
} }
if len(jsonNIDs) == 0 { if len(jsonNIDs) == 0 {
return nil return nil
} }
for i := range jsonNIDs {
d.Cache.EvictFederationQueuedEDU(jsonNIDs[i])
}
if err = d.FederationQueueJSON.DeleteQueueJSON(ctx, txn, jsonNIDs); err != nil { if err = d.FederationQueueJSON.DeleteQueueJSON(ctx, txn, jsonNIDs); err != nil {
return err return err
@ -198,4 +197,14 @@ func (d *Database) DeleteExpiredEDUs(ctx context.Context) error {
return d.FederationQueueEDUs.DeleteExpiredEDUs(ctx, txn, expiredBefore) return d.FederationQueueEDUs.DeleteExpiredEDUs(ctx, txn, expiredBefore)
}) })
if err != nil {
return err
}
for i := range jsonNIDs {
d.Cache.EvictFederationQueuedEDU(jsonNIDs[i])
}
return nil
} }