agent: detect and disable abandoned tasks

[jshearer]

We have a bunch of abandoned tasks ¹. We want to figure out a way to deal with them. In order to do that, we need to:

• Identify when a task is abandoned
• Notify the owner of the task that it's abandoned, and give them some time to remedy the situation
• Disable the abandoned task

Out of scope for this piece of work are tasks that are running but haven't moved any data, dealing with the collections of abandoned tasks, and subsequently deleting the disabled abandoned tasks.

Proposed UX

1. A user publishes a database capture on day 1 of their POC
2. On day 3, they change the firewall rules so that database is no longer reachable
3. The task starts failing, and the control-plane diligently restarts it according to the task failure backoff strategy.
4. On day 3+N (14 maybe?), they get an email saying something like "Your task has been failing for N days. We've tried restarting it Q times without success. It will be disabled in M (maybe 7?) days from now. "
5. On day 3+N+M, the control-plane will create a publication that disables that task.

Definition

A task counts as "abandoned" when:

• It's enabled
• It is a type of entity that has shards (captures, materializations, derivations)
  • Note: This excludes all Dekaf materializations, which is fine because they shouldn't be incurring the same kind of control-plane overhead as actual abandoned tasks do anyway.
• No shard has been PRIMARY since a configurable threshold of days ago. When PRIMARY has never been observed, created_at is used as the fallback, so new tasks aren't flagged until they've existed for the full threshold.

Approach

We'll use the controller-evaluated alert path (same as ShardFailed, AutoDiscoverFailed, BackgroundPublicationFailed). The per-spec controller already does periodic shard listings, so we already have most of the data we need.

One more piece of data is needed:

• last_primary_ts: Option<DateTime<Utc>> on ActivationStatus. Set to now()² whenever aggregate_shard_status() returns Ok during a health check. No new data-plane calls, this piggy-backs on the existing list_task_shards(). Starts as None for existing tasks and updates with the normal health-check cadence.

Then, we'll add a new evaluate_abandoned() function and wire it into the capture, materialization, and collection codepaths after the activation stage. The controller runs the full pipeline on every wake, and broken tasks still wake regularly as e.g shard health checks back off to ~60min ³. Derivations are collections, so they're covered by the collection pipeline. We'll gate with has_task_shards() which returns false for plain collections, Dekaf tasks, disabled tasks etc. I'll call set_alert_firing(TaskAbandoned, ...) or resolve_alert(TaskAbandoned) based on whether the task meets the criteria above.

Concept: Would it also be worth tracking the number of times the controller has restarted a failing task, since it was last PRIMARY? That way we could include in the email some text like "We have attempted to restart your task 826 times since it started failing 14 days ago.". This would also be a useful metric to have in general.

Rollout

We want to roll this out in stages.

Detection only. Fire alerts into alert_history but don't notify anyone. Don't add task_abandoned to default include_alert_types in alert_subscriptions. Deploy with a generous threshold (30 days) and let last_primary_ts accumulate. Monitor alert_history for task_abandoned entries and manually review what gets flagged.

Notification. Once we're confident the detection is flagging the right tasks, add task_abandoned to default include_alert_types, and probably run a migration to opt existing tenants in as well. By this point we'll need a real email template describing what "abandoned" means and what the user should do about it.

Auto-disable. When TaskAbandoned has been firing for longer than ABANDONED_TASK_DISABLE_AFTER (default 7 days), the controller creates a background publication setting shards.disable = true.

• On success, the task is disabled, is_enabled_task() returns false on the next run, the alert resolves, and the controller stops scheduling health checks.
  • Note: Make sure this doesn't send a "your task is no longer abandoned" alert resolution email when we disable lol.
• On failure, the alert stays firing and the controller retries on its next wake.

(Un)answered questions

Data-plane outage: if the data-plane that a task is running on goes down, what happens? Will the controller handle not being able to make shard listings? Will the controller handle not being able to publish the disablement? I imagine as designed, we'll sit there trying over and over to disable these tasks.

• Answer from talking with @psFried: yep, we don't have a good story around data-plane deletions. currently this will get stuck in a loop trying to activate the disablement or deletion unless support manually removes all tasks from a data-plane before removing it.

Existing abandoned tasks: As designed, this will apply to both existing and new abandoned tasks, after the threshold window has passed. That's probably what we want, just calling it out

How do we want to communicate: Do we want an email after X number of days without PRIMARY, then disable after Y? Multiple emails? Do we want a resolution email? etc

Windows: What are we thinking WRT the thresholds here? 14 days to warning email, then another week until disable?

Footnotes

1. How many? ↩

2. Is now() correct? Do we have some other concept of time to worry about here?
   ———
   Answer: The controller uses control_plane.current_time() rather than raw now(). In production this is just Utc::now(), but the indirection exists so that integration tests can use a controllable clock. ↩

3. Is this correct? I want to make sure I/we understand how often this automation will run. I feel like it doesn't need to run more than roughly once per day, but more often also shouldn't hurt.
   ———
   Answer: the controller picks the soonest of the reported wake-at times from the sub-tasks, so evaluate_abandoned() will just report whatever interval it cares about, like 24h, and that will guarantee that it gets woken no later than that. ↩

[jgraettinger]

Thanks for a thoughtful write-up!

Absence of PRIMARY is a conservative and definitive signal that things are not okay, but qualitatively I don't think it's a sufficient one. I've observed many apparently abandoned, repeat-failing tasks which are able to reach PRIMARY but then fail shortly thereafter -- typically due to a terminal error that can only be encountered at runtime, from a query attempt, and isn't or cannot be checked for up-front as part of Validate (usually connectors are running the same health-checks on startup before responding with Opened, than they run in Validate -- the runtime promotes to PRIMARY upon receiving Opened).

Given the notification and CTA we're planning to deliver, I think we should be less conservative w.r.t. what we detect as an "abandoned" task. Perhaps coupling repeat failures with lack of data movement.

[dyaffe]

Just one question from me -- is this the same method that @psFried used to find the initial batch?

[jshearer]

Happy to broaden/extend the definition of what an abandoned task is, but I don't quite understand this part:

Absence of PRIMARY is a conservative and definitive signal that things are not okay [...] given the notification and CTA we're planning to deliver, I think we should be less conservative

I would assume that given the CTA and eventual disablement, we would want to be conservative to avoid false-positives and disabling tasks that aren't abandoned. No?

---

is this the same method that @psFried used to find the initial batch?

I doubt it, this proposal requires tracking the last time a task was PRIMARY, which I don't believe we currently do.

[jshearer]

To broaden the definition to accomodate the scenario you presented, we could use the existing ShardStatusCheck.count field:

flow/crates/models/src/status/activation.rs

Lines 23 to 26 in 1ebe8e6

	pub struct ShardStatusCheck {
	/// The number of checks that have returned ths status
	#[serde(default, skip_serializing_if = "crate::is_u32_zero")]
	pub count: u32,

Instead of updating last_primary_ts whenever we see a PRIMARY status, we could instead update it when we have a PRIMARY status and count > P, where P is some number corresponding to how many times the shard status check ran and observed PRIMARY.

For example, if we picked P=3, the shard would need to stay PRIMARY for ~9 minutes before updating last_primary_ts. P=4 gives you 19 minutes, etc:

flow/crates/agent/src/controllers/activation.rs

Lines 584 to 591 in 1ebe8e6

	// If the status is OK, then backoff much more quickly
	let duration = if current_status == ShardsStatus::Ok {
	match prev_checks {
	0..3 => Duration::minutes(3),
	3..6 => Duration::minutes(10),
	6..10 => Duration::minutes(60),
	_ => max_duration,
	}

[psFried]

I think it's a fair point that a controller can observe a shard in PRIMARY status even though it may still fail shortly after. Though I also agree that it makes sense to start with conservative criteria and gradually expand the definition after.

We do already have a somewhat more robust system for determining task health, in the status summary. In particular, that looks at whether the connector has emitted a ConnectorStatus event after the last restart. We could potentially do the same here, taking the timestamp of the most recent connector status instead of the timestamp of the most recently observed PRIMARY shard status. This could technically still fail to identify tasks that, for example, fail right after writing the status. But the ConnectorStatus is at least intended to serve as a positive health check, so maybe we'd consider it a bug in the connector if that happened?

I'm also not against checking catalog_stats to see whether a task has successfully moved any data. My hesitance there is only because I already worry about the performance of our stats tables, and I'd want to make sure that we don't have controllers querying them too frequently until we get a better handle on it. Perhaps that'd be good as a final check, after we've checked all the other stuff, though.

One other topic that Joseph and I had discussed, just to share the outcome: at least in many cases, abandoned tasks will also fail to publish. Looking at some of the recently disabled tasks, many of them have publication histories with entries similar to this example:

{
  "id": "redacted",
  "created": "2025-07-29T09:48:02.080402932Z",
  "completed": "2026-02-24T14:49:38.042746489Z",
  "detail": "periodic publication",
  "result": {
    "type": "buildFailed",
    "lockFailures": []
  },
  "errors": [
    {
      "catalogName": "redacted",
      "scope": "flow://materialization/redacted/materialize-iceberg",
      "detail": "creating catalog: failed to GET https://glue.ap-south-1.amazonaws.com/iceberg/v1/config?warehouse=redacted: 403 Forbidden (Code: 0, Message: , Type: )"
    }
  ],
  "isTouch": true,
  "count": 3001
}

We'd discussed this earlier, and decided against using publication success as a criteria for what's abandoned. It's possible that publications could still succeed even if the task itself might never actually work. A webhook materialization is a good example, since there's no way for a Validate call to reliably determine whether the actual HTTP requests will fail or not. So our takeaway was that we should focus on whether the task is actually working or not, and ignore publication errors (which will eventually cause the task to stop working if our periodic publication fails for long enough).

[jshearer]

So, signals that will be combined to determine whether a task is abandoned:

• Clearly desired: task is enabled
• Clearly desired: last_primary_ts (tracking the last time the connector has been PRIMARY for >= P polls in a row, equating to some amount of time) is older than some threshold
• Probably desired: the last time the connector emitted a status is older than some threshold
• Possibly desired: the last time the connector stats record data in/out is older than some threshold.
  • This seems smart to me, except for adding more logic dependent on the current stats tables, which both makes migrating harder in the future, and adds more load to already overloaded tables.

[jgraettinger]

My general feedback, without delving too much into specifics, is that:

• It's okay and desired to be more aggressive in heuristics that decide whether a task looks abandoned because we're giving the user plenty of advance notice and an opportunity to remediate or say "hey this is actually not abandoned, you got it wrong!".
  • We should expect that "you got it wrong!" may happen, and it's okay. In fact it's a key product feedback signal.
  • If we weren't giving notice, we would need to be way more conservative: turning off a task that false positives as abandoned without prior notice would be very upsetting.
• If we make this too conservative, we'll definitely keep coming back to this issue out of a desire and need to fix false negatives.
  • Given that, it'd be unfortunate to sink a bunch of implementation time into strategies that seem too-conservative out of the gate. For example, if we know now that PRIMARY is a too-weak signal, why cruft up controller states with it? How is it load bearing?
  • ConnectorStatus could make more sense, but it'd be worth surveying integrations members on whether this is high signal. It has the great advantage of already being in controller state as I understand, so experimentation is cheaper.
    • But, I think a survey of usage would show that it's not meaningfully better than PRIMARY as a signal, and is too conservative.

[jgraettinger]

Other signals that matter for a determination of abandoned:

• Failure rate of a task
• Amount of data moved (catalog stats monthly?)
• Whether the account is paid vs free tier
• Recency of latest user publication
• Presence of an end-to-end data flow

For example, there are a lot of free tier Google Sheets captures that move no data month after month, and aren't even hooked up to a materialization, but don't actually "fail" regularly. These consume an exorbitant amount of API calls that affect shared rate limits and can impact paying customers with real use cases. Can we find an approach that would detect these as abandoned as well?

[jshearer]

Failure rate of a task

So, a task that stays PRIMARY for extended periods of time, but also fails frequently should be considered abandoned? I'm trying to figure out how this meshes with the conditions that were already mentioned: last time the connector stayed PRIMARY for a certain amount of time, and the last time the connector emitted a status message.

---

Amount of data moved

I addressed this here. The risk seems to be twofold: adding more logic dependent on the current stats tables making migrating harder in the future, and also adding more load to already overloaded tables. It's a fairly straightforward query to add all things considered... thoughts on these concerns?

What would the condition be? "Even if PRIMARY, if no data moved for >X days consider abandoned"?

---

Whether the account is paid vs free tier

Do we not disable paid tasks at all/should this whole thing only apply to free tier tasks? Or are you suggesting we have a more generous window of time before we either notify, or disable if you're paid vs free? Also, what about public vs private data-planes? It seems like we might want to notify but not disable on a private data-plane? 🤔

---

Recency of latest user publication

Phil and I talked about this one, and we decided it didn't seem worth it:

It's possible that publications could still succeed even if the task itself might never actually work. A webhook materialization is a good example, since there's no way for a Validate call to reliably determine whether the actual HTTP requests will fail or not. So our takeaway was that we should focus on whether the task is actually working or not, and ignore publication errors (which will eventually cause the task to stop working if our periodic publication fails for long enough).

---

Presence of an end-to-end data flow

This seems meaningful to me... but what's the condition? Do we not disable a task that's not moving data and hasn't been PRIMARY for the defined timeout if it's connected to a collection that's read from by a materialization (and/or is the sourceCapture of a materialization)?

[jshearer]

Johnny, Phil, and I spoke just now and we concluded that all of these conditions can boil down to two separate alert sequences:

• Task failing for extended period of time: fires when a task has had a ShardFailed alert firing for > 30 days. This applies to all tenants, free or paid. In short, "the task has not been able to stay alive for longer than 2 hours in the past N days".
  • ShardFailed triggers upon shard failed log, i.e whenever a task has a failure, and it also triggers when a task is in FAILED state for whatever reason.
  • ShardFailed resolves once the task has been observed in PRIMARY for > 2 hours.
• Task has not moved data for an extended period of time: fires when:
  • A task has no catalog stats in/out for >30 days
  • The task has not been published by a user for > 30 days
  • The task is in a tenant that's in the free tier
  • The task does not have an active ShardFailed alert (in order to avoid getting both of these emails at once for a free tier task that has been failing and hasn't written any data)

Each alert sequence contains two emails:

• A first email that fires as soon as the condition is met (i.e the 30 day threshold is passed), informing you about the condition and what you can do to remedy it, as well as when we plan to disable the task if not remedied (maybe 7 days in the future?)
• A second email that fires when we have disabled the task, and what you can do to get it back up and running

The intent with splitting these up into separate alert types is that we'll likely want to treat these more as marketing nudges to bring users back, rather than harsh warnings that their tasks are bad and they should feel bad.

@dyaffe @jwhartley what feedback do you have? Any meaningful conditions we missed? All of the timelines can be adjusted granularly, i.e per alert type and sequence. Eventually it would be nice to have some copy for these emails, but that's after we lock down exactly what they represent.

[jwhartley]

Overall looks good to me.

Task has not moved data for an extended period of time: fires when:

• The task has not been published by a user for > 30 days
• The task is in a tenant that's in the free tier
  Can you explain the thinking for these two conditions?

User publish: Is the idea that the customer is actively trying to fix the task if there's recent changes? I'd say 14 days is more reasonable than 30 if so. I have a slight concern that some customers will be automatically deploying tasks periodically, thus invalidating this condition, but really they should only be publishing if there's a change.

Free tier: I think notification of prolonged no data movement is a useful signal for paid customers, and disabling under these conditions is unlikely to upset them.

[dyaffe]

• I think we should disable paid tasks
• I'd look at the end-to-end dataflow thing as an improvement vs a V1

I guess my overall feedback would be:

1. Lets try to do a real MVP then add bells and whistles. No need to boil the ocean...simple, one or two metrics and get it working.
2. Before the automation does anything, lets have it send me and the solutions team a list to look over. After that, we can set it live to cause all sorts of havoc.