feature: handle delete operation for IPs

This commit is contained in:
ayurishchev committed 2026-08-23 22:24:55 +03:00
1 parent 4f597505cc
commit f8336740ad
20 files changed
+1083 -5

No files matched your search

+8
View File
@@ -154,3 +154,11 @@ type SubmitIPsResult struct {
Reordered []string
SkippedInProgress []string
}
// DeleteIPsResult categorizes how each address in a DeleteIPs call (or a
// ClearQueue call, which is DeleteIPs given every currently queued address)
// was handled.
type DeleteIPsResult struct {
Deleted []string
NotFound []string
}
+101
View File
@@ -2,6 +2,7 @@ package db
import (
"context"
"database/sql"
"errors"
"path/filepath"
"testing"
@@ -330,3 +331,103 @@ func TestCancelIP(t *testing.T) {
t.Fatalf("expected cancelled + owner cleared, got %+v", got)
}
}
func TestDeleteIP(t *testing.T) {
d, ctx := newTestDB(t)
if err := d.SeedQueue(ctx, []string{"1.2.3.4"}); err != nil {
t.Fatalf("seed queue: %v", err)
}
ip, err := d.GetIPByAddress(ctx, "1.2.3.4")
if err != nil {
t.Fatalf("get ip: %v", err)
}
if err := d.UpsertCheck(ctx, Check{
IPID: ip.ID, IPAddress: ip.IPAddress, AttemptNumber: ip.AttemptNumber,
Source: SourceEgress, CheckType: "https", Target: "https://example.test",
Success: true, CheckedAt: Now(),
}); err != nil {
t.Fatalf("upsert check: %v", err)
}
if err := d.InsertEvent(ctx, Event{SourceType: "control-api", IPID: &ip.ID, EventType: "test_event", OccurredAt: Now()}); err != nil {
t.Fatalf("insert event: %v", err)
}
if err := d.DeleteIP(ctx, ip.ID); err != nil {
t.Fatalf("delete ip: %v", err)
}
if _, err := d.GetIP(ctx, ip.ID); !errors.Is(err, sql.ErrNoRows) {
t.Fatalf("expected row gone, got %v", err)
}
checks, err := d.ListChecksForAttempt(ctx, ip.ID, ip.AttemptNumber)
if err != nil {
t.Fatalf("list checks: %v", err)
}
if len(checks) != 0 {
t.Fatalf("expected checks gone, got %+v", checks)
}
events, err := d.ListEventsForIP(ctx, ip.ID)
if err != nil {
t.Fatalf("list events: %v", err)
}
if len(events) != 0 {
t.Fatalf("expected events gone, got %+v", events)
}
if err := d.DeleteIP(ctx, ip.ID); !errors.Is(err, ErrNotFound) {
t.Fatalf("expected ErrNotFound deleting already-gone ip, got %v", err)
}
}
func TestDeleteIPFreesOwningValidator(t *testing.T) {
d, ctx := newTestDB(t)
if err := d.SeedQueue(ctx, []string{"1.2.3.4"}); err != nil {
t.Fatalf("seed queue: %v", err)
}
if err := d.AdminCreateValidator(ctx, "validator-1", "port-1"); err != nil {
t.Fatalf("create validator: %v", err)
}
claimed, err := d.ClaimNextQueued(ctx, "validator-1", time.Minute)
if err != nil || claimed == nil {
t.Fatalf("claim: item=%+v err=%v", claimed, err)
}
if err := d.DeleteIP(ctx, claimed.ID); err != nil {
t.Fatalf("delete ip: %v", err)
}
v, err := d.GetValidator(ctx, "validator-1")
if err != nil {
t.Fatalf("get validator: %v", err)
}
if v.State != ValidatorIdle || v.CurrentIPID != nil {
t.Fatalf("expected validator freed, got state=%s current_ip=%v", v.State, v.CurrentIPID)
}
}
func TestDeleteIPs(t *testing.T) {
d, ctx := newTestDB(t)
if err := d.SeedQueue(ctx, []string{"1.1.1.1", "2.2.2.2"}); err != nil {
t.Fatalf("seed queue: %v", err)
}
result, err := d.DeleteIPs(ctx, []string{"1.1.1.1", "no-such-address", "2.2.2.2"})
if err != nil {
t.Fatalf("delete ips: %v", err)
}
if len(result.Deleted) != 2 || result.Deleted[0] != "1.1.1.1" || result.Deleted[1] != "2.2.2.2" {
t.Fatalf("expected both real addresses deleted, got %+v", result)
}
if len(result.NotFound) != 1 || result.NotFound[0] != "no-such-address" {
t.Fatalf("expected unknown address in NotFound, got %+v", result)
}
ips, err := d.ListIPs(ctx)
if err != nil {
t.Fatalf("list ips: %v", err)
}
if len(ips) != 0 {
t.Fatalf("expected empty queue, got %+v", ips)
}
}
+84
View File
@@ -335,6 +335,90 @@ func (d *DB) CancelIP(ctx context.Context, ipID int64) error {
return nil
}
// DeleteIP permanently removes an ip_queue row, along with its full check
// and event history, in one transaction — frees the owning validator (if
// any) back to idle first, same as ForceCancel does for the DB side.
// Unlike CancelIP, this leaves nothing behind: the row and its history are
// gone, not marked cancelled. Disassociating a currently-attached floating
// IP is the caller's (orchestrator's) job, same division as CancelIP.
// Returns ErrNotFound if the address is unknown.
func (d *DB) DeleteIP(ctx context.Context, ipID int64) error {
tx, err := d.BeginTx(ctx, nil)
if err != nil {
return err
}
defer tx.Rollback()
if err := deleteIPTx(ctx, tx, ipID); err != nil {
return err
}
return tx.Commit()
}
// DeleteIPs deletes a specific list of addresses in one transaction,
// tolerating unknown addresses the same way SubmitIPs does: each is
// resolved to an id and either deleted (added to Deleted) or, if unknown,
// added to NotFound rather than aborting the whole call. Also the
// implementation behind "clear queue" — call it with every address
// currently in the queue.
func (d *DB) DeleteIPs(ctx context.Context, addresses []string) (DeleteIPsResult, error) {
var result DeleteIPsResult
tx, err := d.BeginTx(ctx, nil)
if err != nil {
return result, err
}
defer tx.Rollback()
for _, addr := range addresses {
var ipID int64
err := tx.QueryRowContext(ctx, `SELECT id FROM ip_queue WHERE ip_address=?`, addr).Scan(&ipID)
if err == sql.ErrNoRows {
result.NotFound = append(result.NotFound, addr)
continue
}
if err != nil {
return result, err
}
if err := deleteIPTx(ctx, tx, ipID); err != nil {
return result, fmt.Errorf("delete %s: %w", addr, err)
}
result.Deleted = append(result.Deleted, addr)
}
if err := tx.Commit(); err != nil {
return result, err
}
return result, nil
}
// deleteIPTx is the shared body of DeleteIP/DeleteIPs: free the owning
// validator, delete dependent checks/events, then the ip_queue row itself
// — the same FK-clearing order DeleteValidator uses for owner_validator_id.
func deleteIPTx(ctx context.Context, tx *sql.Tx, ipID int64) error {
now := timeToDB(Now())
if _, err := tx.ExecContext(ctx, `
UPDATE validators SET state=?, current_ip_id=NULL, updated_at=?
WHERE current_ip_id=?
`, ValidatorIdle, now, ipID); err != nil {
return fmt.Errorf("free owning validator: %w", err)
}
if _, err := tx.ExecContext(ctx, `DELETE FROM checks WHERE ip_id=?`, ipID); err != nil {
return fmt.Errorf("delete checks: %w", err)
}
if _, err := tx.ExecContext(ctx, `DELETE FROM events WHERE ip_id=?`, ipID); err != nil {
return fmt.Errorf("delete events: %w", err)
}
res, err := tx.ExecContext(ctx, `DELETE FROM ip_queue WHERE id=?`, ipID)
if err != nil {
return fmt.Errorf("delete ip_queue row: %w", err)
}
if n, _ := res.RowsAffected(); n == 0 {
return fmt.Errorf("ip_id %d: %w", ipID, ErrNotFound)
}
return nil
}
func (d *DB) SetEgressComplete(ctx context.Context, ipID int64) error {
_, err := d.ExecContext(ctx, `UPDATE ip_queue SET egress_complete=1, updated_at=? WHERE id=?`, timeToDB(Now()), ipID)
return err