package events
import (
"encoding/json"
"errors"
"fmt"
"strings"
"time"
"github.com/nats-io/nats.go"
)
// ForgetRepo erases what the streams hold about a deleted repository:
// the attention events addressed to anybody about it, its event log,
// and its traffic tallies. A repository created later under the same
// name must not inherit them (quarry#294) — an inbox row that opens the
// new repository, a clone count and the accounts behind it that were
// somebody else's.
//
// Messages are matched on the repository they NAME, never on a subject
// prefix alone: repository names may contain dots, so "gone.>" also
// covers "gone.x", which is a different repository.
func (s *Server) ForgetRepo(repo string) error {
if s == nil || repo == "" {
return nil
}
named := func(data []byte) bool {
var e struct {
Repo string `json:"repo"`
}
return json.Unmarshal(data, &e) == nil && e.Repo == repo
}
if err := s.deleteWhere(AttnStream, "attn.>", named); err != nil {
return err
}
if err := s.deleteWhere(VocabStream, "quarry.evt."+repo+".>", named); err != nil {
return err
}
return s.forgetTallies(repo)
}
// forgetTallies purges repo's traffic tallies. A counter body carries
// only the total, so ownership is read off the subject, and repository
// names may contain dots: "quarry.count.a.clone.auth.clone" is account
// "clone" cloning a, and equally repository a.clone.auth's clone total.
// A subject is a's only when no longer name can claim it — an ambiguous
// one is left, because erasing a neighbour's tally is worse than
// keeping one of a's for an account literally named clone or fetch.
func (s *Server) forgetTallies(repo string) error {
head := CounterSubject(repo, "")
info, err := s.js.StreamInfo(CounterStream, &nats.StreamInfoRequest{SubjectsFilter: head + ">"})
if err != nil {
if errors.Is(err, nats.ErrStreamNotFound) {
return nil
}
return fmt.Errorf("events: forget %s tallies: %w", repo, err)
}
for subj := range info.State.Subjects {
rest := strings.Split(strings.TrimPrefix(subj, head), ".")
if !tallyOf(rest) {
continue
}
claimed := false
for k := 1; k < len(rest) && !claimed; k++ {
claimed = tallyOf(rest[k:])
}
if claimed {
continue
}
if err := s.js.PurgeStream(CounterStream, &nats.StreamPurgeRequest{Subject: subj}); err != nil {
return fmt.Errorf("events: forget %s: %w", subj, err)
}
}
return nil
}
// tallyOf reports whether tokens are one tally as cmd/quarry counts
// them: clone or fetch, then nothing, anon, auth or claim and a token,
// or client and two tokens.
func tallyOf(tokens []string) bool {
if len(tokens) == 0 || (tokens[0] != "clone" && tokens[0] != "fetch") {
return false
}
switch len(tokens) {
case 1:
return true
case 2:
return tokens[1] == "anon"
case 3:
return tokens[1] == "auth" || tokens[1] == "claim"
case 4:
return tokens[1] == "client"
}
return false
}
// forgetBatch is how many messages one fetch asks for; forgetIdleLimit
// how many empty fetches in a row a scan tolerates while the consumer
// still reports messages pending.
var (
forgetBatch = 256
forgetIdleLimit = 25
)
// forgetFetch is one fetch of a forget scan; a test stands in a short one.
var forgetFetch = func(sub *nats.Subscription, n int) ([]*nats.Msg, error) {
return sub.Fetch(n, nats.MaxWait(200*time.Millisecond))
}
// deleteWhere deletes every message under filter in stream whose body
// match accepts.
func (s *Server) deleteWhere(stream, filter string, match func([]byte) bool) error {
sub, err := s.js.PullSubscribe(filter, "",
nats.DeliverAll(), nats.AckNone(), nats.BindStream(stream),
nats.InactiveThreshold(30*time.Second))
if err != nil {
if errors.Is(err, nats.ErrStreamNotFound) {
return nil
}
return fmt.Errorf("events: forget in %s: %w", stream, err)
}
defer sub.Unsubscribe()
var doomed []uint64
// A fetch may return fewer messages than asked for with more still
// waiting, so the scan ends on the consumer's own count of what it
// has not delivered, never on a short batch.
idle := 0
for {
msgs, ferr := forgetFetch(sub, forgetBatch)
if ferr != nil && !errors.Is(ferr, nats.ErrTimeout) {
return fmt.Errorf("events: forget in %s: %w", stream, ferr)
}
for _, m := range msgs {
if !match(m.Data) {
continue
}
if meta, merr := m.Metadata(); merr == nil {
doomed = append(doomed, meta.Sequence.Stream)
}
}
info, ierr := sub.ConsumerInfo()
if ierr != nil {
return fmt.Errorf("events: forget in %s: %w", stream, ierr)
}
if info.NumPending == 0 {
break
}
if len(msgs) == 0 {
if idle++; idle >= forgetIdleLimit {
return fmt.Errorf("events: forget in %s: %d messages never delivered", stream, info.NumPending)
}
continue
}
idle = 0
}
for _, seq := range doomed {
if err := s.js.DeleteMsg(stream, seq); err != nil && !errors.Is(err, nats.ErrMsgNotFound) {
return fmt.Errorf("events: forget %s#%d: %w", stream, seq, err)
}
}
return nil
}