~james/core

core/spindle/db/backfill.go -rw-r--r-- 3.1 KiB
452f17b7 — james chore(import): snapshot james.dad.town/core at f70c71c7696a 16 hours ago
                                                                                
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
package db

import (
	"database/sql"
	"time"

	"github.com/bluesky-social/indigo/atproto/syntax"
)

type ParkedRecord struct {
	URI      string
	Rev      string
	Event    []byte
	Attempts int
	RetryAt  time.Time
	Err      string
}

func (d *DB) ParkRecord(p ParkedRecord) error {
	_, err := d.Exec(
		`insert into backfill_parked (uri, rev, event, attempts, retry_at, error)
		 values (?, ?, ?, 0, ?, ?)
		 on conflict(uri) do update set
		     rev      = excluded.rev,
		     event    = excluded.event,
		     attempts = 0,
		     retry_at = excluded.retry_at,
		     error    = excluded.error
		 where excluded.rev >= backfill_parked.rev`,
		p.URI, p.Rev, p.Event, p.RetryAt.Unix(), p.Err,
	)
	return err
}

func (d *DB) DueParkedRecords(before time.Time) ([]ParkedRecord, error) {
	rows, err := d.Query(
		`select uri, rev, event, attempts, retry_at, error from backfill_parked
		 where retry_at <= ? order by retry_at`,
		before.Unix(),
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []ParkedRecord
	for rows.Next() {
		var p ParkedRecord
		var retryAt int64
		if err := rows.Scan(&p.URI, &p.Rev, &p.Event, &p.Attempts, &retryAt, &p.Err); err != nil {
			return nil, err
		}
		p.RetryAt = time.Unix(retryAt, 0)
		out = append(out, p)
	}
	return out, rows.Err()
}

func (d *DB) NextParkedRetry() (time.Time, error) {
	var at sql.NullInt64
	if err := d.QueryRow(`select min(retry_at) from backfill_parked`).Scan(&at); err != nil {
		return time.Time{}, err
	}
	if !at.Valid {
		return time.Time{}, nil
	}
	return time.Unix(at.Int64, 0), nil
}

func (d *DB) RescheduleParked(uri, rev string, attempts int, retryAt time.Time, cause string) error {
	_, err := d.Exec(
		`update backfill_parked set attempts = ?, retry_at = ?, error = ?
		 where uri = ? and rev = ?`,
		attempts, retryAt.Unix(), cause, uri, rev,
	)
	return err
}

func (d *DB) UnparkRecord(uri, rev string) error {
	_, err := d.Exec(`delete from backfill_parked where uri = ? and rev = ?`, uri, rev)
	return err
}

func (d *DB) CountParked() (int, error) {
	var n int
	err := d.QueryRow(`select count(*) from backfill_parked`).Scan(&n)
	return n, err
}

// AddPendingCatchUp keeps the first since, so a did that never backfills still ages out
func (d *DB) AddPendingCatchUp(dids []syntax.DID, since time.Time) error {
	tx, err := d.Begin()
	if err != nil {
		return err
	}
	defer tx.Rollback()
	for _, did := range dids {
		if _, err := tx.Exec(`insert or ignore into backfill_pending (did, since) values (?, ?)`, did, since.Unix()); err != nil {
			return err
		}
	}
	return tx.Commit()
}

func (d *DB) PendingCatchUp() ([]syntax.DID, error) {
	return queryDIDs(d, `select did from backfill_pending order by did`)
}

func (d *DB) DeletePendingCatchUp(dids ...syntax.DID) error {
	tx, err := d.Begin()
	if err != nil {
		return err
	}
	defer tx.Rollback()
	for _, did := range dids {
		if _, err := tx.Exec(`delete from backfill_pending where did = ?`, did); err != nil {
			return err
		}
	}
	return tx.Commit()
}

func (d *DB) DropStalePendingCatchUp(olderThan time.Time) ([]syntax.DID, error) {
	return queryDIDs(d, `delete from backfill_pending where since < ? returning did`, olderThan.Unix())
}