From 16deb65082b340aed5b341cd333dbe645b0d14ad Mon Sep 17 00:00:00 2001 From: Fishcake Date: Tue, 25 Apr 2023 09:37:58 +0900 Subject: [PATCH 1/4] feat(relay):reuse relay connections --- nostr.go | 67 +++++++++++++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 64 insertions(+), 3 deletions(-) diff --git a/nostr.go b/nostr.go index 9c1eca9..f8ea146 100644 --- a/nostr.go +++ b/nostr.go @@ -32,6 +32,18 @@ var nip57Receipt nostr.Event var zapEventSerializedStr string var nip57ReceiptRelays []string +// Relay connections +type RelayConnection struct { + URL string + relay *nostr.Relay + lastUsed time.Time + closeChan chan bool +} + +var relayConnections = make(map[string]*RelayConnection) +var relayConnectionsMutex sync.Mutex +var connectionTimeout = 30 * time.Minute + func Nip57DescriptionHash(zapEventSerialized string) string { hash := sha256.Sum256([]byte(zapEventSerialized)) hashString := hex.EncodeToString(hash[:]) @@ -195,6 +207,56 @@ func GetNostrProfileMetaData(npub string) (nostr.ProfileMetadata, error) { } +func getRelayConnection(url string) (*nostr.Relay, error) { + relayConnectionsMutex.Lock() + defer relayConnectionsMutex.Unlock() + + if relayConn, ok := relayConnections[url]; ok { + relayConn.lastUsed = time.Now() + return relayConn.relay, nil + } + + ctx := context.WithValue(context.Background(), "url", url) + relay, err := nostr.RelayConnect(ctx, url) + if err != nil { + return nil, err + } + + relayConn := &RelayConnection{ + URL: url, + relay: relay, + lastUsed: time.Now(), + closeChan: make(chan bool), + } + relayConnections[url] = relayConn + + go func() { + select { + case <-time.After(connectionTimeout): + relayConnectionsMutex.Lock() + if time.Since(relayConn.lastUsed) >= connectionTimeout { + relay.Close() + delete(relayConnections, url) + } + relayConnectionsMutex.Unlock() + case <-relayConn.closeChan: + } + }() + + return relay, nil +} + +func closeRelayConnection(url string) { + relayConnectionsMutex.Lock() + defer relayConnectionsMutex.Unlock() + + if relayConn, ok := relayConnections[url]; ok { + relayConn.closeChan <- true + relayConn.relay.Close() + delete(relayConnections, url) + } +} + func publishNostrEvent(ev nostr.Event, relays []string) { // Add more relays, remove trailing slashes, and ensure unique relays relays = uniqueSlice(cleanUrls(append(relays, Relays...))) @@ -209,16 +271,15 @@ func publishNostrEvent(ev nostr.Event, relays []string) { go func(url string) { defer wg.Done() - ctx := context.WithValue(context.Background(), "url", url) - relay, err := nostr.RelayConnect(ctx, url) + relay, err := getRelayConnection(url) if err != nil { log.Printf("Error connecting to relay %s: %v", url, err) return } - defer relay.Close() time.Sleep(3 * time.Second) + ctx := context.WithValue(context.Background(), "url", url) status, err := relay.Publish(ctx, ev) if err != nil { log.Printf("Error publishing to relay %s: %v", url, err) From 010ff4ee00ac628252aa0a51bb590d3d8389a12e Mon Sep 17 00:00:00 2001 From: Fishcake Date: Tue, 25 Apr 2023 09:53:22 +0900 Subject: [PATCH 2/4] feat(relays): ignore relays we cannot connect to, with timeout --- nostr.go | 27 +++++++++++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/nostr.go b/nostr.go index f8ea146..fab02bc 100644 --- a/nostr.go +++ b/nostr.go @@ -43,6 +43,9 @@ type RelayConnection struct { var relayConnections = make(map[string]*RelayConnection) var relayConnectionsMutex sync.Mutex var connectionTimeout = 30 * time.Minute +var ignoreRelayDuration = 10 * time.Minute +var ignoredRelays = make(map[string]time.Time) +var ignoredRelaysMutex sync.Mutex func Nip57DescriptionHash(zapEventSerialized string) string { hash := sha256.Sum256([]byte(zapEventSerialized)) @@ -207,7 +210,30 @@ func GetNostrProfileMetaData(npub string) (nostr.ProfileMetadata, error) { } +func ignoreRelay(url string) { + ignoredRelaysMutex.Lock() + defer ignoredRelaysMutex.Unlock() + ignoredRelays[url] = time.Now() +} + +func isRelayIgnored(url string) bool { + ignoredRelaysMutex.Lock() + defer ignoredRelaysMutex.Unlock() + + if t, ok := ignoredRelays[url]; ok { + if time.Since(t) < ignoreRelayDuration { + return true + } + delete(ignoredRelays, url) + } + return false +} + func getRelayConnection(url string) (*nostr.Relay, error) { + if isRelayIgnored(url) { + return nil, fmt.Errorf("relay %s is being ignored", url) + } + relayConnectionsMutex.Lock() defer relayConnectionsMutex.Unlock() @@ -219,6 +245,7 @@ func getRelayConnection(url string) (*nostr.Relay, error) { ctx := context.WithValue(context.Background(), "url", url) relay, err := nostr.RelayConnect(ctx, url) if err != nil { + ignoreRelay(url) return nil, err } From 6b0b54132a6c348b6ba91d2b0fa7449c8cb8edfd Mon Sep 17 00:00:00 2001 From: Fishcake Date: Tue, 25 Apr 2023 10:04:32 +0900 Subject: [PATCH 3/4] fix(relays): also ignore relays that fail to publish, shorten time to ignore --- nostr.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/nostr.go b/nostr.go index fab02bc..04b92dc 100644 --- a/nostr.go +++ b/nostr.go @@ -43,7 +43,7 @@ type RelayConnection struct { var relayConnections = make(map[string]*RelayConnection) var relayConnectionsMutex sync.Mutex var connectionTimeout = 30 * time.Minute -var ignoreRelayDuration = 10 * time.Minute +var ignoreRelayDuration = 5 * time.Minute var ignoredRelays = make(map[string]time.Time) var ignoredRelaysMutex sync.Mutex @@ -309,6 +309,7 @@ func publishNostrEvent(ev nostr.Event, relays []string) { ctx := context.WithValue(context.Background(), "url", url) status, err := relay.Publish(ctx, ev) if err != nil { + ignoreRelay(url) log.Printf("Error publishing to relay %s: %v", url, err) } else { log.Printf("[NOSTR] published to %s: %s", url, status) From ab13d1fbfb168bb9e2294023c31f846abf09a453 Mon Sep 17 00:00:00 2001 From: Fishcake Date: Tue, 25 Apr 2023 13:04:46 +0900 Subject: [PATCH 4/4] fix(relays): handle closed connections --- nostr.go | 58 +++++++++++++++++++++++++++++++++++++++----------------- 1 file changed, 41 insertions(+), 17 deletions(-) diff --git a/nostr.go b/nostr.go index 04b92dc..0ce442b 100644 --- a/nostr.go +++ b/nostr.go @@ -5,7 +5,9 @@ import ( "crypto/sha256" "encoding/hex" "encoding/json" + "errors" "fmt" + "net" "net/http" "strings" "sync" @@ -229,6 +231,16 @@ func isRelayIgnored(url string) bool { return false } +func isBrokenPipeError(err error) bool { + var netErr net.Error + if errors.As(err, &netErr) { + if strings.Contains(netErr.Error(), "write: broken pipe") { + return true + } + } + return false +} + func getRelayConnection(url string) (*nostr.Relay, error) { if isRelayIgnored(url) { return nil, fmt.Errorf("relay %s is being ignored", url) @@ -298,24 +310,36 @@ func publishNostrEvent(ev nostr.Event, relays []string) { go func(url string) { defer wg.Done() - relay, err := getRelayConnection(url) - if err != nil { - log.Printf("Error connecting to relay %s: %v", url, err) - return + var err error + var relay *nostr.Relay + var status nostr.Status + maxRetries := 3 + + for i := 0; i < maxRetries; i++ { + relay, err = getRelayConnection(url) + if err != nil { + log.Printf("Error connecting to relay %s: %v", url, err) + return + } + + time.Sleep(3 * time.Second) + + ctx := context.WithValue(context.Background(), "url", url) + status, err = relay.Publish(ctx, ev) + if err != nil { + log.Printf("Error publishing to relay %s: %v", url, err) + + if isBrokenPipeError(err) { + closeRelayConnection(url) // Close the broken connection + continue // Retry connection and publish + } + } else { + log.Printf("[NOSTR] published to %s: %s", url, status.String()) // Convert the nostr.Status value to a string + break + } + + time.Sleep(3 * time.Second) } - - time.Sleep(3 * time.Second) - - ctx := context.WithValue(context.Background(), "url", url) - status, err := relay.Publish(ctx, ev) - if err != nil { - ignoreRelay(url) - log.Printf("Error publishing to relay %s: %v", url, err) - } else { - log.Printf("[NOSTR] published to %s: %s", url, status) - } - - time.Sleep(3 * time.Second) }(url) }