mirrors/dendrite
1
0
Fork 0
mirror of https://github.com/matrix-org/dendrite.git synced 2025-12-23 14:53:10 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix build & lint

Browse source
This commit is contained in:
Matthew Hodgson 2020-09-11 22:39:50 +01:00
parent a5c0521c3f
commit df29509e7e
7 changed files with 51 additions and 48 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
federationsender/api/api.go
View file

@ -114,7 +114,7 @@ type PerformJoinResponse struct {
type PerformPeekRequest struct { type PerformPeekRequest struct {
RoomID string `json:"room_id"` RoomID string `json:"room_id"`
// The sorted list of servers to try. Servers will be tried sequentially, after de-duplication. // The sorted list of servers to try. Servers will be tried sequentially, after de-duplication.
ServerNames types.ServerNames `json:"server_names"` ServerNames types.ServerNames `json:"server_names"`
} }
type PerformPeekResponse struct { type PerformPeekResponse struct {

21
federationsender/internal/perform.go
View file

@ -186,7 +186,7 @@ func (r *FederationSenderInternalAPI) performJoinUsingServer(
// Check that the send_join response was valid. // Check that the send_join response was valid.
joinCtx := perform.JoinContext(r.federation, r.keyRing) joinCtx := perform.JoinContext(r.federation, r.keyRing)
respState, err := joinCtx.CheckSendJoinResponse( respState, err := joinCtx.CheckSendJoinResponse(
ctx, event, serverName, respMakeJoin, respSendJoin, ctx, event, serverName, respSendJoin,
) )
if err != nil { if err != nil {
return fmt.Errorf("joinCtx.CheckSendJoinResponse: %w", err) return fmt.Errorf("joinCtx.CheckSendJoinResponse: %w", err)
@ -195,10 +195,11 @@ func (r *FederationSenderInternalAPI) performJoinUsingServer(
// If we successfully performed a send_join above then the other // If we successfully performed a send_join above then the other
// server now thinks we're a part of the room. Send the newly // server now thinks we're a part of the room. Send the newly
// returned state to the roomserver to update our local view. // returned state to the roomserver to update our local view.
headeredEvent := event.Headered(respMakeJoin.RoomVersion)
if err = roomserverAPI.SendEventWithState( if err = roomserverAPI.SendEventWithState(
ctx, r.rsAPI, ctx, r.rsAPI,
respState, respState,
event.Headered(respMakeJoin.RoomVersion), &headeredEvent,
nil, nil,
); err != nil { ); err != nil {
return fmt.Errorf("r.producer.SendEventWithState: %w", err) return fmt.Errorf("r.producer.SendEventWithState: %w", err)
@ -293,18 +294,17 @@ func (r *FederationSenderInternalAPI) performPeekUsingServer(
// check whether we're peeking already to try to avoid needlessly // check whether we're peeking already to try to avoid needlessly
// re-peeking on the server. we don't need a transaction for this, // re-peeking on the server. we don't need a transaction for this,
// given this is a nice-to-have. // given this is a nice-to-have.
remotePeek, err := r.db.GetRemotePeek(ctx, roomID, serverName, peekID) remotePeek, err := r.db.GetRemotePeek(ctx, serverName, roomID, peekID)
if err != nil { if err != nil {
return err return err
} }
renewing := false renewing := false
if remotePeek != nil { if remotePeek != nil {
nowMilli := time.Now().UnixNano() / int64(time.Millisecond) nowMilli := time.Now().UnixNano() / int64(time.Millisecond)
if (nowMilli > remotePeek.RenewedTimestamp + remotePeek.RenewalInterval) { if nowMilli > remotePeek.RenewedTimestamp+remotePeek.RenewalInterval {
logrus.Infof("stale remote peek to %s for %s already exists; renewing", serverName, roomID) logrus.Infof("stale remote peek to %s for %s already exists; renewing", serverName, roomID)
renewing = true renewing = true
} } else {
else {
logrus.Infof("live remote peek to %s for %s already exists", serverName, roomID) logrus.Infof("live remote peek to %s for %s already exists", serverName, roomID)
return nil return nil
} }
@ -336,12 +336,11 @@ func (r *FederationSenderInternalAPI) performPeekUsingServer(
// If we've got this far, the remote server is peeking. // If we've got this far, the remote server is peeking.
if renewing { if renewing {
if err = r.db.RenewRemotePeek(ctx, serverName, roomID, respPeek.RenewalInterval); err != nil { if err = r.db.RenewRemotePeek(ctx, serverName, roomID, peekID, respPeek.RenewalInterval); err != nil {
return err return err
} }
} } else {
else { if err = r.db.AddRemotePeek(ctx, serverName, roomID, peekID, respPeek.RenewalInterval); err != nil {
if err = r.db.AddRemotePeek(ctx, serverName, roomID, respPeek.RenewalInterval); err != nil {
return err return err
} }
} }
@ -351,7 +350,7 @@ func (r *FederationSenderInternalAPI) performPeekUsingServer(
if err = roomserverAPI.SendEventWithState( if err = roomserverAPI.SendEventWithState(
ctx, r.rsAPI, ctx, r.rsAPI,
&respState, &respState,
event.Headered(respPeek.RoomVersion), nil, nil, nil,
); err != nil { ); err != nil {
return fmt.Errorf("r.producer.SendEventWithState: %w", err) return fmt.Errorf("r.producer.SendEventWithState: %w", err)
} }

9
federationsender/storage/interface.go
View file

@ -50,12 +50,13 @@ type Database interface {
GetPendingPDUServerNames(ctx context.Context) ([]gomatrixserverlib.ServerName, error) GetPendingPDUServerNames(ctx context.Context) ([]gomatrixserverlib.ServerName, error)
GetPendingEDUServerNames(ctx context.Context) ([]gomatrixserverlib.ServerName, error) GetPendingEDUServerNames(ctx context.Context) ([]gomatrixserverlib.ServerName, error)
// XXX: why don't these have contexts passed in?
AddServerToBlacklist(serverName gomatrixserverlib.ServerName) error AddServerToBlacklist(serverName gomatrixserverlib.ServerName) error
RemoveServerFromBlacklist(serverName gomatrixserverlib.ServerName) error RemoveServerFromBlacklist(serverName gomatrixserverlib.ServerName) error
IsServerBlacklisted(serverName gomatrixserverlib.ServerName) (bool, error) IsServerBlacklisted(serverName gomatrixserverlib.ServerName) (bool, error)
AddRemotePeek(serverName gomatrixserverlib.ServerName, roomID, peekID string, renewalInterval int) error AddRemotePeek(ctx context.Context, serverName gomatrixserverlib.ServerName, roomID, peekID string, renewalInterval int) error
RenewRemotePeek(serverName gomatrixserverlib.ServerName, roomID, peekID string, renewalInterval int) error RenewRemotePeek(ctx context.Context, serverName gomatrixserverlib.ServerName, roomID, peekID string, renewalInterval int) error
GetRemotePeek(serverName gomatrixserverlib.ServerName, roomID, peekID string) (types.RemotePeek, error) GetRemotePeek(ctx context.Context, serverName gomatrixserverlib.ServerName, roomID, peekID string) (*types.RemotePeek, error)
GetRemotePeeks(roomID string) ([]types.RemotePeek, error) GetRemotePeeks(ctx context.Context, roomID string) ([]types.RemotePeek, error)
} }

16
federationsender/storage/shared/storage.go
View file

@ -165,22 +165,22 @@ func (d *Database) IsServerBlacklisted(serverName gomatrixserverlib.ServerName)
return d.FederationSenderBlacklist.SelectBlacklist(context.TODO(), nil, serverName) return d.FederationSenderBlacklist.SelectBlacklist(context.TODO(), nil, serverName)
} }
func (d *Database) AddRemotePeek(serverName gomatrixserverlib.ServerName, roomID, peekID string, renewalInterval int) error { func (d *Database) AddRemotePeek(ctx context.Context, serverName gomatrixserverlib.ServerName, roomID, peekID string, renewalInterval int) error {
return d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error { return d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
return d.FederationSenderRemotePeeks.InsertRemotePeek(context.TODO(), txn, serverName, roomID, peekID, renewalInterval) return d.FederationSenderRemotePeeks.InsertRemotePeek(ctx, txn, serverName, roomID, peekID, renewalInterval)
}) })
} }
func (d *Database) RenewRemotePeek(serverName gomatrixserverlib.ServerName, roomID, peekID string, renewalInterval int) error { func (d *Database) RenewRemotePeek(ctx context.Context, serverName gomatrixserverlib.ServerName, roomID, peekID string, renewalInterval int) error {
return d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error { return d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
return d.FederationSenderRemotePeeks.RenewRemotePeek(context.TODO(), txn, serverName, roomID, peekID, renewalInterval) return d.FederationSenderRemotePeeks.RenewRemotePeek(ctx, txn, serverName, roomID, peekID, renewalInterval)
}) })
} }
func (d *Database) GetRemotePeek(serverName gomatrixserverlib.ServerName, roomID, peekID string) (types.RemotePeek, error) { func (d *Database) GetRemotePeek(ctx context.Context, serverName gomatrixserverlib.ServerName, roomID, peekID string) (*types.RemotePeek, error) {
return d.FederationSenderRemotePeeks.SelectRemotePeek(context.TODO(), serverName, roomID, peekID) return d.FederationSenderRemotePeeks.SelectRemotePeek(ctx, nil, serverName, roomID, peekID)
} }
func (d *Database) GetRemotePeeks(roomID string) ([]types.RemotePeek, error) { func (d *Database) GetRemotePeeks(ctx context.Context, roomID string) ([]types.RemotePeek, error) {
return d.FederationSenderRemotePeeks.SelectRemotePeeks(context.TODO(), roomID) return d.FederationSenderRemotePeeks.SelectRemotePeeks(ctx, nil, roomID)
} }

32
federationsender/storage/sqlite3/remote_peeks_table.go
View file

@ -17,7 +17,10 @@ package sqlite3
import ( import (
"context" "context"
"database/sql" "database/sql"
"time"
"github.com/matrix-org/dendrite/federationsender/types"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/sqlutil" "github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/gomatrixserverlib" "github.com/matrix-org/gomatrixserverlib"
) )
@ -97,7 +100,7 @@ func (s *remotePeeksStatements) InsertRemotePeek(
) (err error) { ) (err error) {
nowMilli := time.Now().UnixNano() / int64(time.Millisecond) nowMilli := time.Now().UnixNano() / int64(time.Millisecond)
stmt := sqlutil.TxStmt(txn, s.insertRemotePeekStmt) stmt := sqlutil.TxStmt(txn, s.insertRemotePeekStmt)
_, err := stmt.ExecContext(ctx, roomID, serverName, peekID, nowMilli, nowMilli, renewalInterval) _, err = stmt.ExecContext(ctx, roomID, serverName, peekID, nowMilli, nowMilli, renewalInterval)
return return
} }
@ -105,31 +108,34 @@ func (s *remotePeeksStatements) RenewRemotePeek(
ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName, roomID, peekID string, renewalInterval int, ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName, roomID, peekID string, renewalInterval int,
) (err error) { ) (err error) {
nowMilli := time.Now().UnixNano() / int64(time.Millisecond) nowMilli := time.Now().UnixNano() / int64(time.Millisecond)
_, err := sqlutil.TxStmt(txn, s.renewRemotePeekStmt).ExecContext(ctx, nowMilli, renewalInterval, roomID, serverName, peekID) _, err = sqlutil.TxStmt(txn, s.renewRemotePeekStmt).ExecContext(ctx, nowMilli, renewalInterval, roomID, serverName, peekID)
return return
} }
func (s *remotePeeksStatements) SelectRemotePeek( func (s *remotePeeksStatements) SelectRemotePeek(
ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName, roomID, peekID string, ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName, roomID, peekID string,
) (remotePeek types.RemotePeek, err error) { ) (*types.RemotePeek, error) {
rows, err := sqlutil.TxStmt(txn, s.selectRemotePeeksStmt).QueryContext(ctx, roomID) rows, err := sqlutil.TxStmt(txn, s.selectRemotePeeksStmt).QueryContext(ctx, roomID)
if err != nil { if err != nil {
return return nil, err
} }
defer internal.CloseAndLogIfError(ctx, rows, "SelectRemotePeek: rows.close() failed") defer internal.CloseAndLogIfError(ctx, rows, "SelectRemotePeek: rows.close() failed")
remotePeek := types.RemotePeek{} remotePeek := types.RemotePeek{}
if err = rows.Scan( err = rows.Scan(
&remotePeek.RoomID, &remotePeek.RoomID,
&remotePeek.ServerName, &remotePeek.ServerName,
&remotePeek.PeekID, &remotePeek.PeekID,
&remotePeek.CreationTimestamp, &remotePeek.CreationTimestamp,
&remotePeek.RenewTimestamp, &remotePeek.RenewedTimestamp,
&remotePeek.RenewalInterval, &remotePeek.RenewalInterval,
); err != nil { )
return if err == sql.ErrNoRows {
return nil, nil
} }
return remotePeek, rows.Err() if err != nil {
return nil, err
}
return &remotePeek, rows.Err()
} }
func (s *remotePeeksStatements) SelectRemotePeeks( func (s *remotePeeksStatements) SelectRemotePeeks(
@ -148,7 +154,7 @@ func (s *remotePeeksStatements) SelectRemotePeeks(
&remotePeek.ServerName, &remotePeek.ServerName,
&remotePeek.PeekID, &remotePeek.PeekID,
&remotePeek.CreationTimestamp, &remotePeek.CreationTimestamp,
&remotePeek.RenewTimestamp, &remotePeek.RenewedTimestamp,
&remotePeek.RenewalInterval, &remotePeek.RenewalInterval,
); err != nil { ); err != nil {
return return
@ -162,13 +168,13 @@ func (s *remotePeeksStatements) SelectRemotePeeks(
func (s *remotePeeksStatements) DeleteRemotePeek( func (s *remotePeeksStatements) DeleteRemotePeek(
ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName, roomID, peekID string, ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName, roomID, peekID string,
) (err error) { ) (err error) {
_, err := sqlutil.TxStmt(txn, s.deleteRemotePeekStmt).ExecContext(ctx, roomID, serverName, peekID) _, err = sqlutil.TxStmt(txn, s.deleteRemotePeekStmt).ExecContext(ctx, roomID, serverName, peekID)
return return
} }
func (s *remotePeeksStatements) DeleteRemotePeeks( func (s *remotePeeksStatements) DeleteRemotePeeks(
ctx context.Context, txn *sql.Tx, roomID string, ctx context.Context, txn *sql.Tx, roomID string,
) (err error) { ) (err error) {
_, err := sqlutil.TxStmt(txn, s.deleteRemotePeeksStmt).ExecContext(ctx, roomID) _, err = sqlutil.TxStmt(txn, s.deleteRemotePeeksStmt).ExecContext(ctx, roomID)
return return
} }

2
federationsender/storage/tables/interface.go
View file

@ -71,7 +71,7 @@ type FederationSenderBlacklist interface {
type FederationSenderRemotePeeks interface { type FederationSenderRemotePeeks interface {
InsertRemotePeek(ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName, roomID, peekID string, renewalInterval int) (err error) InsertRemotePeek(ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName, roomID, peekID string, renewalInterval int) (err error)
RenewRemotePeek(ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName, roomID, peekID string, renewalInterval int) (err error) RenewRemotePeek(ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName, roomID, peekID string, renewalInterval int) (err error)
SelectRemotePeek(ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName, roomID, peekID string) (remotePeek types.RemotePeek, err error) SelectRemotePeek(ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName, roomID, peekID string) (remotePeek *types.RemotePeek, err error)
SelectRemotePeeks(ctx context.Context, txn *sql.Tx, roomID string) (remotePeeks []types.RemotePeek, err error) SelectRemotePeeks(ctx context.Context, txn *sql.Tx, roomID string) (remotePeeks []types.RemotePeek, err error)
DeleteRemotePeek(ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName, roomID, peekID string) (err error) DeleteRemotePeek(ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName, roomID, peekID string) (err error)
DeleteRemotePeeks(ctx context.Context, txn *sql.Tx, roomID string) (err error) DeleteRemotePeeks(ctx context.Context, txn *sql.Tx, roomID string) (err error)

17
federationsender/types/types.go
View file

@ -50,14 +50,11 @@ func (e EventIDMismatchError) Error() string {
) )
} }
// UnixMs is the milliseconds since the Unix epoch
type UnixMs int64
type RemotePeek struct { type RemotePeek struct {
PeekID string PeekID string
RoomID string RoomID string
ServerName gomatrixserverlib.ServerName ServerName gomatrixserverlib.ServerName
CreatedTimestamp UnixMs CreationTimestamp int64
RenewedTimestamp UnixMs RenewedTimestamp int64
RenewalInterval UnixMs RenewalInterval int64
} }