1package daemon23import (4 "context"5 "database/sql"6 "strings"7)89// sweepReplicationState mirrors bind.ws pending-state expiry. Pending Git10// events are removed when their retained deadline passes; promotion removes11// the deadline before this sweep can run.12func (t *Tenant) sweepReplicationState(ctx context.Context, now int64) error {13 var pending []string14 rows, err := t.store.DB().QueryContext(ctx, `SELECT id FROM replication_pending_until WHERE until>0 AND until<=?`, now)15 if err == nil {16 for rows.Next() {17 var id string18 if err := rows.Scan(&id); err != nil {19 rows.Close()20 return err21 }22 pending = append(pending, id)23 }24 if err := rows.Err(); err != nil {25 rows.Close()26 return err27 }28 rows.Close()29 }30 if t.git != nil && len(pending) > 0 {31 if err := t.git.ExpirePending(ctx, pending); err != nil {32 return err33 }34 }35 if t.git != nil {36 rows, err := t.store.DB().QueryContext(ctx, `SELECT repo,ref FROM replication_pr_refs WHERE until>0 AND until<=?`, now)37 if err == nil {38 for rows.Next() {39 var repo, ref string40 if err := rows.Scan(&repo, &ref); err != nil {41 rows.Close()42 return err43 }44 var pruneErr error45 if strings.HasPrefix(repo, "pr:") || strings.HasPrefix(repo, "30617:") {46 pruneErr = t.git.DeletePRRef(ctx, repo, ref)47 } else {48 pruneErr = t.git.PruneRef(ctx, repo, ref)49 }50 if pruneErr != nil {51 rows.Close()52 return pruneErr53 }54 }55 if err := rows.Err(); err != nil {56 rows.Close()57 return err58 }59 rows.Close()60 }61 }62 return t.store.WithTx(ctx, func(tx *sql.Tx) error {63 if _, err := tx.ExecContext(ctx, `DELETE FROM events WHERE id IN (SELECT id FROM replication_pending_until WHERE until>0 AND until<=?)`, now); err != nil && !isMissingTable(err) {64 return err65 }66 for _, query := range []string{67 `DELETE FROM pending_events WHERE id IN (SELECT id FROM replication_pending_until WHERE until>0 AND until<=?)`,68 `DELETE FROM replication_pending_until WHERE until>0 AND until<=?`,69 `DELETE FROM replication_pr_refs WHERE until>0 AND until<=?`,70 } {71 if _, err := tx.ExecContext(ctx, query, now); err != nil && !isMissingTable(err) {72 return err73 }74 }75 return nil76 })77}7879func isMissingTable(err error) bool {80 return err != nil && (strings.Contains(err.Error(), "no such table") || strings.Contains(err.Error(), "no such column"))81}